{"record":{"id":"d8303d85ec572a94","repo":"apache/beam","slug":"buffer-counter-not-initialized-for-uuid","errorCode":null,"errorMessage":"Buffer counter not initialized for UUID: ","messagePattern":"Buffer counter 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":237,"sourceCode":"    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;\n\n    // Setup the wrapped DoFn\n    DoFnInvokers.invokerFor(syncFn)\n        .invokeSetup(\n            new DoFnInvoker.BaseArgumentProvider<InputT, OutputT>() {\n              @Override\n              public PipelineOptions pipelineOptions() {\n                return options;","sourceCodeStart":219,"sourceCodeEnd":255,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/AsyncWrapper.java#L219-L255","documentation":"getItemsInBuffer fetches the AtomicInteger backpressure counter for this UUID, created during @Setup. If absent, the wrapper's buffer accounting state is missing and an IllegalStateException is thrown. Called by many internal paths (currentBuffer, scheduleIfRoom, logBackpressure, isEmpty, getItemsInBufferCount), so any use of the wrapper before setup fails.","triggerScenarios":"Any processElement/finishBundle-adjacent call (isEmpty, scheduleIfRoom, logBackpressure, getItemsInBufferCount) executing before @Setup initialized itemsInBuffer, or after teardown.","commonSituations":"Testing the DoFn without DoFnTester; runner misconfiguration that skips DoFn lifecycle; wrong instance/uuid wiring after serialization.","solutions":["Ensure @Setup runs exactly once before element processing (use DoFnTester in tests)","Do not call buffer-related helpers from user code on an un-set-up instance","Report a runner bug if the exception occurs in a real pipeline run with a stock runner"],"exampleFix":"// before\nwrapper.isEmpty(); // called before setup in a unit test\n// after\nDoFnTester.of(wrapper).processBundle(record); // lifecycle handled","handlingStrategy":"try-catch","validationCode":"DoFnInvokers.invokeSetupForTesting(wrapper); // before any isEmpty/scheduleIfRoom call","typeGuard":null,"tryCatchPattern":"try {\n  boolean empty = wrapper.isEmpty();\n} catch (IllegalStateException e) {\n  if (e.getMessage().startsWith(\"Buffer counter not initialized\")) {\n    DoFnInvokers.invokeSetupForTesting(wrapper); // initialize and retry\n  } else throw e;\n}","preventionTips":["Initialize wrapper state exactly once via @Setup before any buffer API use","In unit tests, prefer DoFnTester.processBundle over direct helper calls","File a runner bug if this occurs under a stock runner in production"],"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"}