Fixes #26685

Main Issue: #26685

Motivation

A v5 send that fails because its client is closing can leave the caller's future pending forever. ScalableTopicProducer.finish() hands the failure to the client's completion executor, and PulsarClientImpl.shutdown() stops that executor with shutdownNow(), which drops the tasks still queued on it. A completion queued in the window between the client entering Closing and the executor shutdown is lost. The existing RejectedExecutionException fallback only covers completions submitted after the executor is gone.

This is what made V5ProducerSegmentGoneTest#closingTheClientWhileSendsAreInFlightFailsThemWithoutRecreatingProducers time out on a loaded CI runner (twice in a row, including the TestNG retry): the test keeps issuing sends while the client closes, and on that runner some completions were still queued when the executor was shut down. See #26685. Found while running the CI for #26672 on my fork, whose change does not touch the v5 client; #26684 came out of the same CI runs.

Modifications

• ScalableTopicProducer.finish(): complete the future in place when the client is closing (client.v4Client().isClosed(), the same check isShuttingDown() already uses), the same way it already does once the executor rejects the task. No other behaviour changes; results that arrive while the client is open are still delivered on the completion executor.
• New V5ProducerClosingCompletionTest: holds the client's only completion executor (callbackThreads(1)), closes the client on another thread, issues a send once the client reports closing and asserts that the send fails with AlreadyClosedException instead of never completing. Without the fix the future never completes (3/3 runs: TimeoutException after 5 s); with it the test passes (3/3).

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• V5ProducerClosingCompletionTest#aSendFailedWhileTheClientIsClosingCompletesEvenIfItsCompletionWasQueued fails on master and passes with this change.
• V5ProducerSegmentGoneTest and V5ProducerBackpressureTest still pass (12 tests across the three classes), plus checkstyle and spotless for pulsar-client-v5 and the broker tests.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

While the client is closing, a failed send's future is completed on the thread that failed it instead of on the completion executor. That is the thread finish() already uses once the executor is gone.

A question for reviewers

The check uses the underlying client's state, client.v4Client().isClosed(), because that is what this class already does in isShuttingDown() (and DagWatchClient does the same). The alternative I considered is for PulsarClientV5 to record that closing has begun itself, a flag set in close(), closeAsync() and shutdown() and exposed to the package as isClosed(), so that the v5 layer stops reading the v4 client's state; isShuttingDown() would then use it too. I have that variant ready and can switch if you would rather the v5 client own this. I kept the smaller change here so the bug fix does not carry a design decision with it.

Documentation

• doc-not-needed

Matching PR in forked repository

PR in forked repository: namest504#6