[fix][common] Handle synchronous failures from future suppliers

[Radiancebobo]

Motivation

Some asynchronous helper methods accept a Supplier<CompletableFuture<T>> and assume the supplier always returns a future. However, a supplier can also fail synchronously before creating the future.

Before this change, those synchronous failures could bypass the expected asynchronous error handling path:

• FutureUtil.Sequencer.sequential could throw directly instead of returning a failed future.
• FutureUtil.composeAsync could throw inside the executor task and leave the returned future incomplete.
• Futures.executeWithRetry in managed-ledger could throw before entering the retry state machine, so a retryable transient failure from the operation setup path would not be retried.

These helpers should preserve the CompletableFuture contract: callers should receive a completed or exceptionally completed future, rather than having synchronous exceptions escape or leave the returned future hanging.

Modifications

This change updates the future supplier handling paths to convert synchronous supplier failures into failed futures:

• Added a private helper in FutureUtil to safely invoke Supplier<CompletableFuture<T>>.
• Updated FutureUtil.Sequencer.sequential to use the safe supplier invocation path.
• Updated FutureUtil.composeAsync to complete the returned future exceptionally when the supplier fails synchronously.
• Updated managed-ledger Futures.executeWithRetry to route synchronous supplier failures through the existing retry logic.
• Added null-future handling so suppliers that return null complete exceptionally with NullPointerException instead of causing an unchecked synchronous failure.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added FutureUtilTest.testSequencerReturnsFailedFutureWhenTaskThrowsSynchronously
• Added FutureUtilTest.testComposeAsyncReturnsFailedFutureWhenSupplierThrowsSynchronously
• Added FuturesTest.testExecuteWithRetryHandlesSynchronousFailure

Local verification:

• ./gradlew :pulsar-common:test --tests org.apache.pulsar.common.util.FutureUtilTest -PtestRetryCount=0 --rerun-tasks
• ./gradlew :managed-ledger:test --tests org.apache.bookkeeper.mledger.util.FuturesTest -PtestRetryCount=0 --rerun-tasks

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

[Radiancebobo]

There's a similar PR #25931

[Radiancebobo]

/pulsarbot rerun

[Radiancebobo]

I have executed these and made sure there are no issues.

./gradlew :pulsar-common:test :managed-ledger:test --rerun-tasks -PtestRetryCount=0 \
  --tests org.apache.pulsar.common.util.FutureUtilTest \
  --tests org.apache.bookkeeper.mledger.util.FuturesTest

./gradlew :pulsar-common:spotlessCheck :pulsar-common:checkstyleMain :pulsar-common:checkstyleTest

[Radiancebobo]

@lhotari Could you please do another review for me?

[lhotari]

Sorry for the long wait on your re-review request — that was five weeks, and the delay was mine.

I checked all five of my earlier points against 99405831 and they are genuinely addressed: supplySafely is public with the exact signature I suggested and has javadoc; the null-return message is the one I proposed, verbatim; the javadoc says "is null, throws, or returns null"; and composeAsync now reads supplySafely(futureSupplier).whenComplete(...) with no intermediate variable. CI is green, and I confirmed your new tests actually execute (the OTHER unit group runs :pulsar-common:test and :managed-ledger:test without a group filter).

Worth stating plainly what this fixes, because it is more than tidying. In composeAsync, a supplier that threw — or returned null — did so inside the executor task, so the exception went to the executor's uncaught handler and the returned future was never completed. Its only two callers are LeaderElectionImpl:308 and LockManagerImpl:121, both wrapped in sequencer.sequential(...), so a stranded future there would also wedge that sequencer's chain — leader election and distributed lock handling. Sequencer.sequential could likewise throw synchronously out of a CompletableFuture-returning method, which the project rules forbid. Both are real, and the new tests pin them.

I also verified the CODING.md edit is a genuine correction rather than a loss: checkArgumentAsync exists nowhere in the repository — on master it appears only in that one doc line. Replacing it with supplySafely fixes a dangling reference.

One comment below is worth acting on and is a one-line change in a file you already touch. The other is an optional style nit.

Two things deliberately not raised as problems with this PR, recorded so they are not lost:

• FutureUtil.unwrapCompletionException returns null for a CompletionException/ExecutionException whose cause is null, and Futures.executeWithRetry:83 would then NPE inside its callback and leave resultFuture pending forever. Real mechanism, but I could not find any in-tree way to produce such a wrapper — every construction site wraps a caught exception or an ExecutionException.getCause() from CompletableFuture.get, and the JDK never produces a null cause there. A one-line null-guard in unwrapCompletionException would close the corner globally, but that is hardening of an existing helper, not a defect of this change.
• The helper is a good one, and there are more call sites with exactly the shape it fixes — a bare supplier.get() invoked inside a callback or scheduled task, where a synchronous throw strands the outer future. Concretely: NamespaceResources.runWithMarkDeleteAsync:395 (partitioned-topic delete — a throw there also leaves the topic marked-deleted in metadata), ConsumerImpl:2502 (retry/DLQ producer creation), and MetadataCacheImpl:388 (the BadVersion retry path). All pre-existing and out of scope here; a follow-up sweep would be worthwhile.

[lhotari]

Sorry for the wait — done now, and the delay was on me rather than anything missing from your side. All five of my earlier points check out at 99405831, and I have left one comment worth acting on (a one-line @Test on the neighbouring testSequencer, which turns out never to run) plus one optional style note.

[Radiancebobo]

Sorry for the wait — done now, and the delay was on me rather than anything missing from your side. All five of my earlier points check out at 99405831, and I have left one comment worth acting on (a one-line @Test on the neighbouring testSequencer, which turns out never to run) plus one optional style note.

Thank you for the thorough review and for catching the dormant test.

I’ve added @test to testSequencer(), so it now executes the
thenCompose(...supplySafely(...)) branch and the allowExceptionBreakChain
behavior. I also verified locally that all 13 tests in FutureUtilTest pass.

Could you please take another look when you have a chance?

Thanks again!