apache/beam · error · IllegalStateException

Received data after inbound data receiver is done for instru

Error message

Received data after inbound data receiver is done for instruction %s and transform %s.

What it means

multiplexElements throws IllegalStateException when data arrives for a (instructionId, transformId) whose inbound endpoint is already marked done. The endpoint was terminated (e.g. bundle finished or was aborted) and the harness is still streaming data for it.

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java:223

      Iterator<Elements.Data> dataElements, Iterator<BeamFnApi.Elements.Timers> timerElements)
      throws Exception {
    while (dataElements.hasNext()) {
      // We're careful to avoid references to the full data while processing, allowing the input
      // stream to possibly cleanup memory as it advances.
      InputStream inputStream;
      EndpointStatus<DataEndpoint<?>> endpoint;
      boolean isLast;
      {
        Elements.Data data = dataElements.next();
        isLast = data.getIsLast();
        endpoint = transformIdToDataEndpoint.get(data.getTransformId());
        if (endpoint == null) {
          throw new IllegalStateException(
              String.format(
                  "Unable to find inbound data receiver for instruction %s and transform %s.",
                  data.getInstructionId(), data.getTransformId()));
        } else if (endpoint.isDone) {
          throw new IllegalStateException(
              String.format(
                  "Received data after inbound data receiver is done for instruction %s and transform %s.",
                  data.getInstructionId(), data.getTransformId()));
        }
        inputStream = data.getData().newInput();
      }
      Coder<Object> coder = (Coder<Object>) endpoint.endpoint.getCoder();
      FnDataReceiver<Object> receiver = (FnDataReceiver<Object>) endpoint.endpoint.getReceiver();
      while (inputStream.available() > 0) {
        receiver.accept(coder.decode(inputStream));
      }
      if (isLast) {
        endpoint.isDone = true;
        numEndpointsThatAreIncomplete -= 1;
      }
    }

    while (timerElements.hasNext()) {

View on GitHub (pinned to 12126d8942)

Solutions

1. Stop sending data for a transform once its endpoint is done; check endpoint state before sending.
2. Verify bundle completion signaling (isLast handling) so the harness halts streaming at the correct point.
3. Catch IllegalStateException in awaitCompletion and treat the bundle as failed; check for harness/runner version mismatches if frequent.

Example fix

// before
receiver.accept(elements); // may race with endpoint close
// after
if (!endpoint.isDone) {
  receiver.accept(elements);
}

if (endpoint.isDone) { throw ... } // check endpoint state before sending

try { observer.awaitCompletion(); } catch (IllegalStateException e) { failBundle(instructionId, e); }

When it happens

Trigger: Harness continues sending Elements.Data for a transform after the endpoint was closed/done — late messages, duplicate isLast batches, or racing shutdown during awaitCompletion.

Common situations: Network-layer buffering delivers data after bundle completion; harness/runner race at bundle teardown; retrying a send after the endpoint closed.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• Received timer after inbound timer receiver is done for inst
• Instruction id was poisoned
• Unregistering consumer which was not registered.
• PoisonedException
• UnsupportedOperationException

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/2e55db9613d03f53. Report an issue: GitHub.