{"record":{"id":"23747e6b0f63400a","repo":"apache/beam","slug":"unable-to-find-inbound-data-receiver-for-instruction-s-and","errorCode":null,"errorMessage":"Unable to find inbound data receiver for instruction %s and transform %s.","messagePattern":"Unable to find inbound data receiver 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":218,"sourceCode":"  /**\n   * Dispatches the data and timers from the elements to corresponding receivers. Returns true if\n   * all the endpoints are done after elements dispatching.\n   */\n  public boolean multiplexElements(\n      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;","sourceCodeStart":200,"sourceCodeEnd":236,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataInboundObserver.java#L200-L236","documentation":"multiplexElements looks up the transform id of an inbound Elements.Data in the instruction's endpoint map and throws IllegalStateException when no inbound data receiver is registered for that (instructionId, transformId) pair. The runner/driver must register endpoints for every transform it expects data from before processing completes.","triggerScenarios":"Receiving an Elements.Data whose transformId is absent from transformIdToDataEndpoint during awaitCompletion — e.g. the transform id was never registered via the API, the bundle was set up with different transform ids, or a stale/mismatched instruction's data arrives.","commonSituations":"Harness/runner version mismatch on transform id naming; pipeline graph changed between sends; data for an instruction whose consumer registration was skipped after poisoning.","solutions":["Register data endpoints (via the API's addBeamFnData.../processBundle instruction setup) for every transform id the harness will send.","Log and inspect the offending instructionId/transformId to find which registration was missed.","Ensure instruction ids are not reused across bundles so stale data maps to the correct endpoint set.","Catch IllegalStateException in awaitCompletion and fail the bundle with a clear runner-side error."],"exampleFix":"// before\nobserver.awaitCompletion(); // ISE: unknown transform id\n// after\napi.addBeamFnDataEndpoint(instructionId, transformId, coder, receiver); // register before consuming\nobserver.awaitCompletion();","handlingStrategy":"try-catch","validationCode":"boolean registered = endpointMap.containsKey(data.getTransformId()); if (!registered) { /* register before awaitCompletion */ }","typeGuard":null,"tryCatchPattern":"try { observer.awaitCompletion(); } catch (IllegalStateException e) { failBundle(instructionId, e); }","preventionTips":["Register every transform's data endpoint during bundle setup","Log unknown instructionId/transformId pairs","Keep transform-id naming consistent across harness/runner versions"],"tags":["java","beam-fn","transform-id","receiver-registration"],"backgroundTag":"resource-not-found","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-15T02:17:10.978Z"}