apache/beam · error · IllegalStateException

Thread pool not initialized for UUID:

Error message

Thread pool not initialized for UUID:

What it means

getThreadPool looks up this DoFn instance's ExecutorService in a shared per-UUID map that is populated during @Setup. If the lookup returns null it throws IllegalStateException, meaning the wrapper is being used before setup ran or after teardown. This is an internal invariant: user code should never call it directly, so hitting it indicates a lifecycle misuse or runner bug.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java:220

            ? maxItemsToBuffer
            : Math.max(parallelism * 2, DEFAULT_MIN_BUFFER_CAPACITY);
    this.timeout = (timeout != null) ? timeout : Duration.standardSeconds(DEFAULT_TIMEOUT_SEC);
    this.maxWaitTime =
        (maxWaitTime != null) ? maxWaitTime : Duration.millis(DEFAULT_MAX_WAIT_TIME_MS);
    this.idFn =
        (idFn != null)
            ? idFn
            : (SerializableFunction<InputT, Object>)
                input -> java.util.Objects.requireNonNull(input);
    this.useThreadPool = useThreadPool;
    this.uuid = UUID.randomUUID().toString();
    this.toProcessSpec = (coder != null) ? StateSpecs.bag(coder) : StateSpecs.bag();
  }

  private ExecutorService getThreadPool() {
    ExecutorService threadPool = pool.get(uuid);
    if (threadPool == null) {
      throw new IllegalStateException("Thread pool not initialized for UUID: " + uuid);
    }
    return threadPool;
  }

  @SuppressWarnings("unchecked")
  private ConcurrentHashMap<Object, InFlightElement<OutputT>> getProcessingElements() {
    ConcurrentHashMap<Object, InFlightElement<?>> elements = processingElements.get(uuid);
    if (elements == null) {
      throw new IllegalStateException("Processing elements map not initialized for UUID: " + uuid);
    }
    return (ConcurrentHashMap<Object, InFlightElement<OutputT>>) (ConcurrentHashMap<?, ?>) elements;
  }

  private AtomicInteger getItemsInBuffer() {
    AtomicInteger buffer = itemsInBuffer.get(uuid);
    if (buffer == null) {
      throw new IllegalStateException("Buffer counter not initialized for UUID: " + uuid);
    }

View on GitHub (pinned to 12126d8942)

Solutions

  1. Ensure the runner (or DoFnTester/DoFnInvokers.invokeSetup) invokes @Setup before any processElement calls
  2. In tests, run the DoFn through DoFnTester or TestPipeline rather than calling methods directly
  3. Verify the same DoFn instance that was set up is the one processing elements (no instance swapping)

Example fix

// before
doFn.processElement(context); // direct call in test
// after
DoFnTester<InputT, OutputT> tester = DoFnTester.of(doFn);
tester.processBundle(elements); // runs @Setup first
Defensive patterns

Strategy: try-catch

Validate before calling

// In tests: ensure setup before use
DoFnInvokers.invokeSetupForTesting(doFn);

Try / catch

try {
  doFn.processElement(ctx);
} catch (IllegalStateException e) {
  if (e.getMessage().startsWith("Thread pool not initialized")) {
    DoFnInvokers.invokeSetupForTesting(doFn); // then retry
  } else throw e;
}

Prevention

When it happens

Trigger: Invoking the wrapped async function (via executor -> getThreadPool) when @Setup was not called, e.g. unit-testing the DoFn directly without running DoFnTester/setup, or calling after @Teardown.

Common situations: Hand-rolled test harnesses that call processElement without setup; runner lifecycle bugs or reuse of a torn-down DoFn instance.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

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