apache/beam · error · java.util.concurrent.CompletionException

CompletionException

Error message

CompletionException

What it means

MoreFutures.supplyAsync wraps any InterruptedException thrown by the supplier into java.util.concurrent.CompletionException (with the interrupt re-signaled on the thread). The future completes exceptionally with CompletionException, so callers joining/getting see this exception.

Solutions

1. Catch CompletionException from future.get()/join() and inspect getCause() for InterruptedException
2. Avoid interruptible blocking work in async suppliers, or handle interruption gracefully
3. Ensure the executor is not shut down while work is expected to complete
4. Restore/propagate the interrupt flag if you catch InterruptedException yourself

Example fix

// before
String v = moreFuture.join();
// after
try { String v = moreFuture.join(); }
catch (CompletionException e) {
  if (e.getCause() instanceof InterruptedException) { Thread.currentThread().interrupt(); }
  throw e;
}

try { v = future.get(); } catch (ExecutionException e) { if (e.getCause() instanceof InterruptedException) { Thread.currentThread().interrupt(); } throw e; }

When it happens

Trigger: A supplier passed to MoreFutures.supplyAsync throws InterruptedException while executing on the supplied ExecutorService (e.g. blocked I/O interrupted on shutdown).

Common situations: Executor shutdownNow() interrupting in-flight async work; pipeline teardown cancelling futures; blocking calls inside async suppliers interrupted by timeouts.

Related errors

• Not implemented
• Unexpected null schema for
• Unexpected timeout waiting for element future to resolve…
• 2xx codes should not be exceptions. Got status code
• A 'datagen' table requires either 'rows-per-second' (for…

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/4c719608a74e113f. Report an issue: GitHub.

Appendix: source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/util/MoreFutures.java:105

   */
  public static boolean isCancelled(CompletionStage<?> future) {
    return future.toCompletableFuture().isCancelled();
  }

  /**
   * Like {@link CompletableFuture#supplyAsync(Supplier)} but for {@link ThrowingSupplier}.
   *
   * <p>If the {@link ThrowingSupplier} throws an exception, the future completes exceptionally.
   */
  public static <T> CompletionStage<T> supplyAsync(
      ThrowingSupplier<T> supplier, ExecutorService executorService) {
    return CompletableFuture.supplyAsync(
        () -> {
          try {
            return supplier.get();
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new CompletionException(e);
          } catch (OutOfMemoryError oom) {
            throw oom;
          } catch (Throwable t) {
            throw new CompletionException(t);
          }
        },
        executorService);
  }

  /**
   * Shorthand for {@link #supplyAsync(ThrowingSupplier, ExecutorService)} using {@link
   * ForkJoinPool#commonPool()}.
   */
  public static <T> CompletionStage<T> supplyAsync(ThrowingSupplier<T> supplier) {
    return supplyAsync(supplier, ForkJoinPool.commonPool());
  }

  /**

View on GitHub (pinned to 12126d8942)