apache/beam · error · IllegalStateException

Buffer counter not initialized for UUID:

Error message

Buffer counter not initialized for UUID:

What it means

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.

Source

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

    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;

    // Setup the wrapped DoFn
    DoFnInvokers.invokerFor(syncFn)
        .invokeSetup(
            new DoFnInvoker.BaseArgumentProvider<InputT, OutputT>() {
              @Override
              public PipelineOptions pipelineOptions() {
                return options;

View on GitHub (pinned to 12126d8942)

Solutions

  1. Ensure @Setup runs exactly once before element processing (use DoFnTester in tests)
  2. Do not call buffer-related helpers from user code on an un-set-up instance
  3. Report a runner bug if the exception occurs in a real pipeline run with a stock runner

Example fix

// before
wrapper.isEmpty(); // called before setup in a unit test
// after
DoFnTester.of(wrapper).processBundle(record); // lifecycle handled
Defensive patterns

Strategy: try-catch

Validate before calling

DoFnInvokers.invokeSetupForTesting(wrapper); // before any isEmpty/scheduleIfRoom call

Try / catch

try {
  boolean empty = wrapper.isEmpty();
} catch (IllegalStateException e) {
  if (e.getMessage().startsWith("Buffer counter not initialized")) {
    DoFnInvokers.invokeSetupForTesting(wrapper); // initialize and retry
  } else throw e;
}

Prevention

When it happens

Trigger: Any processElement/finishBundle-adjacent call (isEmpty, scheduleIfRoom, logBackpressure, getItemsInBufferCount) executing before @Setup initialized itemsInBuffer, or after teardown.

Common situations: Testing the DoFn without DoFnTester; runner misconfiguration that skips DoFn lifecycle; wrong instance/uuid wiring after serialization.

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