apache/beam · error · IllegalStateException

Processing elements map not initialized for UUID:

Error message

Processing elements map not initialized for UUID:

What it means

getProcessingElements reads the per-UUID ConcurrentHashMap tracking in-flight elements, populated during @Setup. A null entry means the wrapper's per-instance state was never initialized or already torn down, so an IllegalStateException is thrown. It surfaces when processing elements (via activeElements) on an incorrectly-lifecycle-managed instance.

Source

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

                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);
    }
    return buffer;
  }

  // Setup is called by the runner exactly once on each worker node when this DoFn is initialized.
  // It is responsible for setting up the wrapped synchronous DoFn
  // and initializing the shared JVM-wide thread pool and registries.
  @Setup
  public void setup(PipelineOptions options) {
    this.pipelineOptions = options;

View on GitHub (pinned to 12126d8942)

Solutions

Invoke @Setup (DoFnInvokers.invokeSetup or DoFnTester) before processing elements
Do not reuse a DoFn instance after @Teardown has run
Check that the uuid used for the map matches the instance's own uuid (no cross-instance copying of fields)

Example fix

// before
new AsyncWrapperInstance().activeElements(); // state never set up
// after
DoFnTester.of(wrapper).processBundle(records); // setup invoked by tester

Defensive patterns

Strategy: try-catch

Validate before calling

// Ensure setup ran:
// DoFnInvokers.invokeSetupForTesting(wrapper); before calling activeElements()

Try / catch

try {
  activeElements();
} catch (IllegalStateException e) {
  if (e.getMessage().startsWith("Processing elements map not initialized")) {
    DoFnInvokers.invokeSetupForTesting(wrapper);
  } else throw e;
}

Prevention

Run all wrapper interactions through DoFnTester which handles setup
Do not copy wrapper state fields between instances (uuid-keyed maps must match)
Avoid deserialization paths that drop transient state without re-running setup

When it happens

Trigger: Calling activeElements (or anything that reads the processing-elements map) before @Setup populated state or after @Teardown cleared it; using a fresh DoFn instance without setup in a custom harness.

Common situations: Direct method invocation in unit tests; runner bugs that skip setup; serializing/deserializing the DoFn and losing transient state maps.

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/5a7f9d7c6a26022e. Report an issue: GitHub.