Add structured concurrency wrapper for CompletableFuture by cconstable · Pull Request #2939 · temporalio/sdk-java

cconstable · 2026-07-02T22:06:57Z

What was changed

Add a small, nonblocking structured concurrency-like wrapper for CompletableFuture. Loosely inspired by Java 25 StructuredTaskScope and Swift TaskGroup.

Allows for us to write this:

CompletableFuture<List<String>> result =
    TaskScope.withScope( // 1. create a scope
        scope -> {
            scope.attach(startWork()); // 2. add tasks to the scope
            scope.attach(startMoreWork())
                .thenApply(value -> doDownstreamWork(value));

            return scope.awaitAll(); // 3. collect results
        });

result.get();        // 4. get results a usual
// or
result.cancel(true); // 5. cancel (propagates to all child tasks)

See the README for more details.

Why?

I'm doing extstore work and fan-in / fan-out + proper cancellation + error handling boilerplate with raw CompletableFuture is tedious and error prone. Scoped task creation gives us correctness guarantees and better ergonomics. I noticed other disparate abstractions solving similar problems and provided a general solution.

Checklist

I wrote and generated a ton of tests for this and also ran some "stress" tests locally. Since this is just a wrapper on top of CompletableFuture and we aren't creating/using our own executors or threading abstractions it's not as scary of a thing as it looks.

cconstable · 2026-07-02T22:30:10Z

+          Throwable unwrapped = unwrap(error);
+          if (unwrapped instanceof CancellationException) {
+            next.completeExceptionally(unwrapped);
+            return;
+          }


cancellation is an error but it's not something we can "recover from" so we should just immediately surface cancellations (skip the user supplied transform)

cconstable · 2026-07-03T01:51:58Z

+
+/** Token that allows asynchronous code to observe cancellation requests. */
+@Experimental
+public interface CancellationToken {


the only "public" part of this PR

meant to serve as a general purpose cancellation token

can replace existing bespoke tokens e.g. ActivityCancellationToken

I think we should replace ActivityCancellationToken as part of this PR - it's so new probably no one is using it yet and still experimental. I suppose the only issue is that CancellationException is not ActivityCanceledException which the activity token needs to throw.

Maybe we can parameterize this over exception type?

cconstable · 2026-07-03T02:21:18Z

+   * Attaches an existing asynchronous task to this scope. Scope cancellation requests cancellation
+   * on the attached future.
+   */
+  <U> TaskChain<U> attach(CompletableFuture<U> future);


This is the main usage point:

scope.attach(someAsyncWork());

It lifts a CompletableFuture into the structured scope. It's the bridge from the unstructured CF world into the structured one. We return a TaskChain rather than a CompletableFuture so callers can chain downstream work (via map, recover, etc) without leaving the scope. It also does not expose blocking operations and foot-guns like exceptionally that break cancellation.

cconstable · 2026-07-03T02:43:01Z

+  private ListUtils() {}
+
+  /** Concatenates a list of lists into a single list, preserving order. */
+  public static <T> List<T> flatten(List<? extends List<? extends T>> lists) {


This isn't used in this PR but it's a natural part of the using this abstraction:

scope.awaitAll(ListUtils::flatten);

It's also being used in the branch I have this relies on this branch.

…ableFuture.

Sushisource

The readme example is very motivating.

I don't have any particularly strong comments. Just a few minor things / questions

Sushisource · 2026-07-07T17:52:21Z

+
+/** Token that allows asynchronous code to observe cancellation requests. */
+@Experimental
+public interface CancellationToken {


I think we should replace ActivityCancellationToken as part of this PR - it's so new probably no one is using it yet and still experimental. I suppose the only issue is that CancellationException is not ActivityCanceledException which the activity token needs to throw.

Maybe we can parameterize this over exception type?

Sushisource · 2026-07-07T17:57:12Z

+   * <p>Code waiting on external work can attach a callback to this future to abort in-flight
+   * requests when cancellation is requested.
+   */
+  default CompletableFuture<Void> getCancellationFuture() {


I think probably this should also throw, like was done with the activity version (or, at least, we should have some way for it to work that way)

Sushisource · 2026-07-07T17:59:59Z

+   * Runs a side effect when this task settles. Unlike {@code CompletableFuture.whenComplete}, the
+   * callback receives the unwrapped throwable ({@code null} on success).
+   */
+  AsyncTask<T> whenSettled(BiConsumer<? super T, ? super Throwable> cb);


Can only one of these parameters be non-null? IE: Either it completed T or threw?

If so I think a separate whenSettledExceptionally method might make more sense, but I don't feel super strongly about that.

Sushisource · 2026-07-07T18:06:15Z

+  public <U> AsyncTask<U> attach(CompletableFuture<U> future) {
+    Objects.requireNonNull(future, "future");


Can use @Nonnull annotation too to help IDEs

Sushisource · 2026-07-07T18:11:01Z

+   *
+   * <p>This method does not close the scope. Scope lifetime remains caller-owned.
+   */
+  <R> CompletableFuture<R> awaitAll(Supplier<R> resultSupplier);


Not sure I get what the point of this is. Seems like the no-param version and the transformer version are the ones that matter. This seems like super thin sugar of the no-param version?

Sushisource · 2026-07-07T18:12:03Z

+   * Non-blocking wait that completes with each collected task's settled outcome in collection
+   * order.


Suggested change

* Non-blocking wait that completes with each collected task's settled outcome in collection

* order.

* Non-blocking wait that completes with each collected task's settled outcome in completion

* order.

Is this true? Unclear to me what collection order means

Sushisource · 2026-07-07T18:16:55Z

+  public T get() {
+    if (status == Status.SUCCESS) return value;
+    throw new IllegalStateException("not a success: " + status, cause);
+  }


Might make more sense to just return an Optional here?

Sushisource · 2026-07-07T18:18:53Z

+
+The equivalent code using `CompletableFuture` directly has to rebuild the same ownership and cancellation rules by hand:
+
+```java


Hahaha Jesus Christ lol

cconstable marked this pull request as ready for review July 2, 2026 22:14

cconstable requested a review from a team as a code owner July 2, 2026 22:14

cconstable commented Jul 2, 2026

View reviewed changes

cconstable force-pushed the structured-concurrency branch from b0455e7 to 6371f6d Compare July 3, 2026 01:49

cconstable commented Jul 3, 2026

View reviewed changes

cconstable added 3 commits July 6, 2026 14:51

add minimal, non-blocking, structured concurrency wrapper for Complet…

a676951

…ableFuture.

narrow ListUtils types, add tests, comments.

a6c47eb

formatting fixes.

16a7629

cconstable force-pushed the structured-concurrency branch from 16ea50d to 16a7629 Compare July 6, 2026 18:51

Sushisource reviewed Jul 7, 2026

View reviewed changes

		public <U> AsyncTask<U> attach(CompletableFuture<U> future) {
		Objects.requireNonNull(future, "future");

		* Non-blocking wait that completes with each collected task's settled outcome in collection
		* order.


		The equivalent code using `CompletableFuture` directly has to rebuild the same ownership and cancellation rules by hand:

		```java

Uh oh!

Conversation

cconstable commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What was changed

Why?

Checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Sushisource left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

cconstable commented Jul 2, 2026 •

edited

Loading