{"record":{"id":"d2f7a439804c810a","repo":"apache/beam","slug":"poisonedexception","errorCode":null,"errorMessage":"PoisonedException","messagePattern":"PoisonedException","errorType":"exception","errorClass":"PoisonedException","httpStatus":null,"severity":"info","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java","lineNumber":274,"sourceCode":"        }\n        for (BeamFnApi.Elements.Timers timers : value.getTimersList()) {\n          if (instructionId.equals(timers.getInstructionId())) {\n            builder.addTimers(timers);\n          }\n        }\n        forwardToConsumerForInstructionId(instructionId, builder.build());\n      }\n    }\n\n    private void forwardToConsumerForInstructionId(String instructionId, BeamFnApi.Elements value) {\n      CloseableFnDataReceiver<BeamFnApi.Elements> consumer;\n      try {\n        CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> consumerFuture =\n            receivers.computeIfAbsent(\n                instructionId,\n                (unused) -> {\n                  if (poisonedInstructionIds.getIfPresent(instructionId) != null) {\n                    throw new PoisonedException();\n                  }\n                  LOG.debug(\n                      \"Received data for instruction {} without consumer ready. \"\n                          + \"Waiting for consumer to be registered.\",\n                      instructionId);\n                  return new CompletableFuture<>();\n                });\n        // The consumer may not be registered until the bundle processor is fully constructed so we\n        // conservatively set\n        // a high timeout.  Poisoning will prevent this for occurring for consumers that will not be\n        // registered.\n        consumer = consumerFuture.get(3, TimeUnit.HOURS);\n        /*\n         * TODO: On failure we should fail any bundles that were impacted eagerly\n         * instead of relying on the Runner harness to do all the failure handling.\n         */\n      } catch (TimeoutException e) {\n        LOG.error(","sourceCodeStart":256,"sourceCodeEnd":292,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java#L256-L292","documentation":"While inserting a placeholder future in forwardToConsumerForInstructionId, the computeIfAbsent callback detects the instruction id was poisoned and throws PoisonedException (a sentinel exception). onNext uses this to discard inbound data for an instruction that previously failed, treating it as ignorable rather than a data-plane error.","triggerScenarios":"gRPC inbound onNext delivers BeamFnApi.Elements for an instruction id that was poisoned via poison(instructionId) after a processing failure, and no receiver future exists for it.","commonSituations":"SDK harness sends trailing/last elements for a bundle the runner already failed; network-level retries redeliver data for an aborted instruction.","solutions":["No user action usually needed: PoisonedException is caught internally and the data is ignored by design.","If seen repeatedly, check for SDK/runner version mismatches causing bundles to be failed while data is still in flight.","Ensure process-bundle failures are communicated promptly (fail bundle) so the harness stops streaming data for that instruction."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try { multiplexer.onNext(elements); } catch (PoisonedException e) { /* expected: ignore data for failed instruction */ }","preventionTips":["Fail bundles promptly so the harness stops streaming","Keep harness and runner versions aligned","Treat data after poisoning as ignorable by contract"],"tags":["java","beam-fn","grpc-multiplexer","poisoned-instruction"],"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-15T02:17:10.978Z"}