{"record":{"id":"2405f87f618d2c06","repo":"apache/beam","slug":"thread-pool-not-initialized-for-uuid","errorCode":null,"errorMessage":"Thread pool not initialized for UUID: ","messagePattern":"Thread pool 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":220,"sourceCode":"            ? maxItemsToBuffer\n            : Math.max(parallelism * 2, DEFAULT_MIN_BUFFER_CAPACITY);\n    this.timeout = (timeout != null) ? timeout : Duration.standardSeconds(DEFAULT_TIMEOUT_SEC);\n    this.maxWaitTime =\n        (maxWaitTime != null) ? maxWaitTime : Duration.millis(DEFAULT_MAX_WAIT_TIME_MS);\n    this.idFn =\n        (idFn != null)\n            ? idFn\n            : (SerializableFunction<InputT, Object>)\n                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    }","sourceCodeStart":202,"sourceCodeEnd":238,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java#L202-L238","documentation":"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.","triggerScenarios":"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.","commonSituations":"Hand-rolled test harnesses that call processElement without setup; runner lifecycle bugs or reuse of a torn-down DoFn instance.","solutions":["Ensure the runner (or DoFnTester/DoFnInvokers.invokeSetup) invokes @Setup before any processElement calls","In tests, run the DoFn through DoFnTester or TestPipeline rather than calling methods directly","Verify the same DoFn instance that was set up is the one processing elements (no instance swapping)"],"exampleFix":"// before\ndoFn.processElement(context); // direct call in test\n// after\nDoFnTester<InputT, OutputT> tester = DoFnTester.of(doFn);\ntester.processBundle(elements); // runs @Setup first","handlingStrategy":"try-catch","validationCode":"// In tests: ensure setup before use\nDoFnInvokers.invokeSetupForTesting(doFn);","typeGuard":null,"tryCatchPattern":"try {\n  doFn.processElement(ctx);\n} catch (IllegalStateException e) {\n  if (e.getMessage().startsWith(\"Thread pool not initialized\")) {\n    DoFnInvokers.invokeSetupForTesting(doFn); // then retry\n  } else throw e;\n}","preventionTips":["Always drive DoFns via DoFnTester or TestPipeline in unit tests","Never invoke DoFn lifecycle methods manually on raw instances","Keep a one-to-one mapping between setup/teardown and instance usage"],"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"}