{"record":{"id":"2e55db9613d03f53","repo":"apache/beam","slug":"received-data-after-inbound-data-receiver-is-done-for","errorCode":null,"errorMessage":"Received data after inbound data receiver is done for instruction %s and transform %s.","messagePattern":"Received data after inbound data receiver is done for instruction (.+?) and transform (.+?)\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java","lineNumber":223,"sourceCode":"      Iterator<Elements.Data> dataElements, Iterator<BeamFnApi.Elements.Timers> timerElements)\n      throws Exception {\n    while (dataElements.hasNext()) {\n      // We're careful to avoid references to the full data while processing, allowing the input\n      // stream to possibly cleanup memory as it advances.\n      InputStream inputStream;\n      EndpointStatus<DataEndpoint<?>> endpoint;\n      boolean isLast;\n      {\n        Elements.Data data = dataElements.next();\n        isLast = data.getIsLast();\n        endpoint = transformIdToDataEndpoint.get(data.getTransformId());\n        if (endpoint == null) {\n          throw new IllegalStateException(\n              String.format(\n                  \"Unable to find inbound data receiver for instruction %s and transform %s.\",\n                  data.getInstructionId(), data.getTransformId()));\n        } else if (endpoint.isDone) {\n          throw new IllegalStateException(\n              String.format(\n                  \"Received data after inbound data receiver is done for instruction %s and transform %s.\",\n                  data.getInstructionId(), data.getTransformId()));\n        }\n        inputStream = data.getData().newInput();\n      }\n      Coder<Object> coder = (Coder<Object>) endpoint.endpoint.getCoder();\n      FnDataReceiver<Object> receiver = (FnDataReceiver<Object>) endpoint.endpoint.getReceiver();\n      while (inputStream.available() > 0) {\n        receiver.accept(coder.decode(inputStream));\n      }\n      if (isLast) {\n        endpoint.isDone = true;\n        numEndpointsThatAreIncomplete -= 1;\n      }\n    }\n\n    while (timerElements.hasNext()) {","sourceCodeStart":205,"sourceCodeEnd":241,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java#L205-L241","documentation":"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.","triggerScenarios":"Harness continues sending Elements.Data for a transform after the endpoint was closed/done — late messages, duplicate isLast batches, or racing shutdown during awaitCompletion.","commonSituations":"Network-layer buffering delivers data after bundle completion; harness/runner race at bundle teardown; retrying a send after the endpoint closed.","solutions":["Stop sending data for a transform once its endpoint is done; check endpoint state before sending.","Verify bundle completion signaling (isLast handling) so the harness halts streaming at the correct point.","Catch IllegalStateException in awaitCompletion and treat the bundle as failed; check for harness/runner version mismatches if frequent."],"exampleFix":"// before\nreceiver.accept(elements); // may race with endpoint close\n// after\nif (!endpoint.isDone) {\n  receiver.accept(elements);\n}","handlingStrategy":"try-catch","validationCode":"if (endpoint.isDone) { throw ... } // check endpoint state before sending","typeGuard":null,"tryCatchPattern":"try { observer.awaitCompletion(); } catch (IllegalStateException e) { failBundle(instructionId, e); }","preventionTips":["Check endpoint done state before accepting more data","Honor isLast semantics exactly","Avoid sending after bundle completion; treat shutdown races as fatal-to-bundle"],"tags":["java","beam-fn","race-condition","endpoint-closed"],"backgroundTag":"invalid-state-transition","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-15T07:17:15.476Z"}