{"record":{"id":"5a7f9d7c6a26022e","repo":"apache/beam","slug":"processing-elements-map-not-initialized-for-uuid","errorCode":null,"errorMessage":"Processing elements map not initialized for UUID: ","messagePattern":"Processing elements map not initialized for UUID: ","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java","lineNumber":229,"sourceCode":"                input -> java.util.Objects.requireNonNull(input);\n    this.useThreadPool = useThreadPool;\n    this.uuid = UUID.randomUUID().toString();\n    this.toProcessSpec = (coder != null) ? StateSpecs.bag(coder) : StateSpecs.bag();\n  }\n\n  private ExecutorService getThreadPool() {\n    ExecutorService threadPool = pool.get(uuid);\n    if (threadPool == null) {\n      throw new IllegalStateException(\"Thread pool not initialized for UUID: \" + uuid);\n    }\n    return threadPool;\n  }\n\n  @SuppressWarnings(\"unchecked\")\n  private ConcurrentHashMap<Object, InFlightElement<OutputT>> getProcessingElements() {\n    ConcurrentHashMap<Object, InFlightElement<?>> elements = processingElements.get(uuid);\n    if (elements == null) {\n      throw new IllegalStateException(\"Processing elements map not initialized for UUID: \" + uuid);\n    }\n    return (ConcurrentHashMap<Object, InFlightElement<OutputT>>) (ConcurrentHashMap<?, ?>) elements;\n  }\n\n  private AtomicInteger getItemsInBuffer() {\n    AtomicInteger buffer = itemsInBuffer.get(uuid);\n    if (buffer == null) {\n      throw new IllegalStateException(\"Buffer counter not initialized for UUID: \" + uuid);\n    }\n    return buffer;\n  }\n\n  // Setup is called by the runner exactly once on each worker node when this DoFn is initialized.\n  // It is responsible for setting up the wrapped synchronous DoFn\n  // and initializing the shared JVM-wide thread pool and registries.\n  @Setup\n  public void setup(PipelineOptions options) {\n    this.pipelineOptions = options;","sourceCodeStart":211,"sourceCodeEnd":247,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java#L211-L247","documentation":"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.","triggerScenarios":"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.","commonSituations":"Direct method invocation in unit tests; runner bugs that skip setup; serializing/deserializing the DoFn and losing transient state maps.","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)"],"exampleFix":"// before\nnew AsyncWrapperInstance().activeElements(); // state never set up\n// after\nDoFnTester.of(wrapper).processBundle(records); // setup invoked by tester","handlingStrategy":"try-catch","validationCode":"// Ensure setup ran:\n// DoFnInvokers.invokeSetupForTesting(wrapper); before calling activeElements()","typeGuard":null,"tryCatchPattern":"try {\n  activeElements();\n} catch (IllegalStateException e) {\n  if (e.getMessage().startsWith(\"Processing elements map not initialized\")) {\n    DoFnInvokers.invokeSetupForTesting(wrapper);\n  } else throw e;\n}","preventionTips":["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"],"tags":["java","apache-beam","lifecycle","internal-state"],"backgroundTag":"internal-invariant-violation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}