You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "Radiancebobo (via GitHub)" <gi...@apache.org> on 2026/06/05 03:56:26 UTC
[PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Radiancebobo opened a new pull request, #25939:
URL: https://github.com/apache/pulsar/pull/25939
### 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`
- `git diff --check`
- `git diff --cached --check`
### Does this pull request potentially affect one of the following parts:
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
*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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3361925786
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -238,11 +251,13 @@ public synchronized CompletableFuture<T> sequential(Supplier<CompletableFuture<T
if (sequencerFuture.isCompletedExceptionally() && allowExceptionBreakChain) {
return sequencerFuture;
}
- return sequencerFuture = newTask.get();
+ return sequencerFuture = getFutureSafely(newTask, "Expected Supplier should not return null");
Review Comment:
> fix the compilation issues after the renaming
Sorry about that! I applied the changes directly on GitHub and missed the other parts that needed to be updated in sync. My local environment isn't updated yet, but I'll get on it and fix everything later today. I'll be more careful next time. Thanks for your patience!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-4627965341
There's a similar PR https://github.com/apache/pulsar/pull/25931
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3361724105
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -238,11 +251,13 @@ public synchronized CompletableFuture<T> sequential(Supplier<CompletableFuture<T
if (sequencerFuture.isCompletedExceptionally() && allowExceptionBreakChain) {
return sequencerFuture;
}
- return sequencerFuture = newTask.get();
+ return sequencerFuture = getFutureSafely(newTask, "Expected Supplier should not return null");
Review Comment:
We'd need to update the AGENTS.md/CONTRIBUTING.md to make it more explicit that this step is necessary before more changes are pushed to a PR:
```
# Lint / verify (license headers, formatting, checkstyle) — run before pushing
./gradlew rat spotlessCheck checkstyleMain checkstyleTest
```
https://github.com/apache/pulsar/blob/master/CONTRIBUTING.md#building
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-4633423909
/pulsarbot rerun
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-4633463490
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #25939:
URL: https://github.com/apache/pulsar/pull/25939
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3888912652
##########
pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java:
##########
@@ -252,4 +254,88 @@ public void testSequencer() {
Assert.assertEquals(list3.get(i), (Integer) i);
}
}
-}
\ No newline at end of file
+
+ @Test
+ public void testSequencerReturnsFailedFutureWhenTaskThrowsSynchronously() throws Exception {
Review Comment:
Thank you for the thorough review and for catching this missing @Test annotation.
I added @Test to testSequencer, so the thenCompose(...supplySafely(...))
branch and allowExceptionBreakChain behavior are now actually exercised.
Verified locally with:
./gradlew :pulsar-common:test --tests org.apache.pulsar.common.util.FutureUtilTest
All 13 tests passed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3364000425
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -200,6 +200,30 @@ public static <T> CompletableFuture<T> failedFuture(Throwable t) {
return future;
}
+ /**
+ * Invokes a supplier that is expected to return a {@link CompletableFuture}, converting synchronous failures into
+ * a failed future.
+ *
+ * If the supplier itself is {@code null}, the returned future is completed exceptionally with a
Review Comment:
is null or returns null
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-5467755308
> 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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3361095133
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -200,6 +200,20 @@ public static <T> CompletableFuture<T> failedFuture(Throwable t) {
return future;
}
+ private static <T> CompletableFuture<T> getFutureSafely(Supplier<CompletableFuture<T>> supplier,
+ String nullFutureMessage) {
Review Comment:
This method could be made public after renaming and simplifying the signature. It's an overkill to add a separate message for the NullPointerException.
```suggestion
public static <T> CompletableFuture<T> supplySafely(Supplier<CompletableFuture<T>> supplier) {
```
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -200,6 +200,20 @@ public static <T> CompletableFuture<T> failedFuture(Throwable t) {
return future;
}
+ private static <T> CompletableFuture<T> getFutureSafely(Supplier<CompletableFuture<T>> supplier,
+ String nullFutureMessage) {
+ CompletableFuture<T> future;
+ try {
+ future = supplier.get();
+ } catch (Throwable t) {
+ return failedFuture(t);
+ }
+ if (future == null) {
+ return failedFuture(new NullPointerException(nullFutureMessage));
Review Comment:
Instead of using an explicity NPE message, something like this would be cover what is needed.
```suggestion
return failedFuture(new NullPointerException("The given supplier returned null, supplier=" + supplier));
```
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -200,6 +200,20 @@ public static <T> CompletableFuture<T> failedFuture(Throwable t) {
return future;
}
+ private static <T> CompletableFuture<T> getFutureSafely(Supplier<CompletableFuture<T>> supplier,
+ String nullFutureMessage) {
Review Comment:
while making this `public`, add proper javadoc
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-5648271328
Thanks for the update. I took another look; the annotation is in place and the focused tests pass.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-5453169178
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3881000655
##########
pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java:
##########
@@ -252,4 +254,88 @@ public void testSequencer() {
Assert.assertEquals(list3.get(i), (Integer) i);
}
}
-}
\ No newline at end of file
+
+ @Test
+ public void testSequencerReturnsFailedFutureWhenTaskThrowsSynchronously() throws Exception {
+ FutureUtil.Sequencer<String> sequencer = FutureUtil.Sequencer.create();
+
+ CompletableFuture<String> future = sequencer.sequential(() -> {
+ throw new IllegalStateException("sync fail");
+ });
+
+ try {
Review Comment:
**[QUALITY] optional: the new exception-path tests could use AssertJ per CODING.md**
Entirely optional, and take it or leave it. `CODING.md:103` says to prefer AssertJ assertions over TestNG ones, and the five new exception-path tests use the `try { … fail(…) } catch (ExecutionException e) { assertTrue(…) }` shape. The AssertJ equivalent is a line shorter and cannot be written in a way that silently passes:
```java
assertThatThrownBy(() -> future.get(2, TimeUnit.SECONDS))
.hasCauseInstanceOf(IllegalStateException.class)
.hasRootCauseMessage("sync fail");
```
Against that: this file is uniformly TestNG-style already, so matching its neighbours is a perfectly reasonable choice, and every one of your blocks does include the `fail(...)`, so none of them can pass by accident. Not a blocker either way.
##########
pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java:
##########
@@ -252,4 +254,88 @@ public void testSequencer() {
Assert.assertEquals(list3.get(i), (Integer) i);
}
}
-}
\ No newline at end of file
+
+ @Test
+ public void testSequencerReturnsFailedFutureWhenTaskThrowsSynchronously() throws Exception {
Review Comment:
**[QUALITY] the only test covering the two sequencer branches this PR rewrites has no @Test and never runs**
This new test is good, but it only reaches `sequential()`'s `isDone()` branch — a fresh sequencer, so `sequencerFuture` is already complete and the call goes to `FutureUtil.java:267`.
The two branches you also rewrote — `thenCompose(__ -> supplySafely(newTask))` at `FutureUtil.java:270-271` — are covered only by `testSequencer` at `FutureUtilTest.java:188`, and **that method never runs**: it has no `@Test`, and the class has no class-level annotation (every other method here is annotated individually — 48, 62, 79, 89, 106, 147, 258, 275, 284, 296, 312, 324). There is no `@Factory` in the file and no TestNG suite XML selecting it by name, and the Gradle setup uses `useTestNG` with no suite file, so only annotated methods execute.
It is dormant on master too (line 186 there), so this is not something you introduced. It matters here because it is also the only coverage of `allowExceptionBreakChain` anywhere, which means the green CI on this PR says nothing about either branch you changed.
Adding `@Test` is one line, in a file this PR already touches. Its suppliers never throw and never return `null`, so `supplySafely` is transparent to it and it should pass as-is — worth running once locally to confirm before pushing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3888920574
##########
pulsar-common/src/test/java/org/apache/pulsar/common/util/FutureUtilTest.java:
##########
@@ -252,4 +254,88 @@ public void testSequencer() {
Assert.assertEquals(list3.get(i), (Integer) i);
}
}
-}
\ No newline at end of file
+
+ @Test
+ public void testSequencerReturnsFailedFutureWhenTaskThrowsSynchronously() throws Exception {
+ FutureUtil.Sequencer<String> sequencer = FutureUtil.Sequencer.create();
+
+ CompletableFuture<String> future = sequencer.sequential(() -> {
+ throw new IllegalStateException("sync fail");
+ });
+
+ try {
Review Comment:
Thanks for the suggestion. I kept the TestNG assertions for consistency with the existing tests in FutureUtilTest. The fail(...) calls ensure that the tests cannot pass silently when no exception is thrown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "Radiancebobo (via GitHub)" <gi...@apache.org>.
Radiancebobo commented on PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#issuecomment-5043255878
@lhotari Could you please do another review for me?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3364006415
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -298,13 +322,16 @@ public static <T> CompletableFuture<T> addTimeoutHandling(CompletableFuture<T> f
}
final CompletableFuture<T> future = new CompletableFuture<>();
try {
- executor.execute(() -> futureSupplier.get().whenComplete((result, error) -> {
- if (error != null) {
- future.completeExceptionally(error);
- return;
- }
- future.complete(result);
- }));
+ executor.execute(() -> {
+ CompletableFuture<T> supplierFuture = supplySafely(futureSupplier);
+ supplierFuture.whenComplete((result, error) -> {
Review Comment:
this reads better if it would be simply `supplySafely(futureSupplier).whenComplete` without the intermediate variable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][common] Handle synchronous failures from future suppliers [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #25939:
URL: https://github.com/apache/pulsar/pull/25939#discussion_r3361696935
##########
pulsar-common/src/main/java/org/apache/pulsar/common/util/FutureUtil.java:
##########
@@ -238,11 +251,13 @@ public synchronized CompletableFuture<T> sequential(Supplier<CompletableFuture<T
if (sequencerFuture.isCompletedExceptionally() && allowExceptionBreakChain) {
return sequencerFuture;
}
- return sequencerFuture = newTask.get();
+ return sequencerFuture = getFutureSafely(newTask, "Expected Supplier should not return null");
Review Comment:
fix the compilation issues after the renaming
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org