{"record":{"id":"b2596a5ba39cf7a9","repo":"apache/beam","slug":"instruction-id-was-poisoned","errorCode":null,"errorMessage":"Instruction id was poisoned","messagePattern":"Instruction id was poisoned","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java","lineNumber":127,"sourceCode":"   * single instruction id.\n   *\n   * <p>The caller must either {@link #unregisterConsumer unregister the consumer} when all messages\n   * have been processed or {@link #poisonInstructionId(String) poison the instruction} if messages\n   * for the instruction should be dropped.\n   */\n  public void registerConsumer(\n      String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {\n    receivers.compute(\n        instructionId,\n        (unused, existing) -> {\n          if (existing != null) {\n            if (!existing.complete(receiver)) {\n              throw new IllegalArgumentException(\"Instruction id was registered twice\");\n            }\n            return existing;\n          }\n          if (poisonedInstructionIds.getIfPresent(instructionId) != null) {\n            throw new IllegalArgumentException(\"Instruction id was poisoned\");\n          }\n          return CompletableFuture.completedFuture(receiver);\n        });\n  }\n\n  /** Unregisters a previously registered consumer. */\n  public void unregisterConsumer(String instructionId) {\n    @Nullable CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture =\n        receivers.remove(instructionId);\n    if (receiverFuture != null && !receiverFuture.isDone()) {\n      // The future must have been inserted by the inbound observer since registerConsumer completes\n      // the future.\n      throw new IllegalArgumentException(\"Unregistering consumer which was not registered.\");\n    }\n  }\n\n  /**\n   * Poisons an instruction id.","sourceCodeStart":109,"sourceCodeEnd":145,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java#L109-L145","documentation":"BeamFnDataGrpcMultiplexer.registerConsumer rejects an instruction id that has been poisoned (marking a failed/aborted instruction). Once poisoned, the id is remembered in an expiring cache so any future registration or late inbound data for it fails fast instead of silently dropping traffic.","triggerScenarios":"Calling registerConsumer(instructionId, receiver) after the instruction was poisoned via poison(instructionId), typically following a processing failure on the instruction.","commonSituations":"Runner-side code retries a failed instruction with the same id; a stale process bundle is re-registered after a pipeline failure; test harnesses reuse instruction ids across scenarios.","solutions":["Use a fresh, unique instruction id when retrying after failure instead of reusing the poisoned one.","Check whether the instruction failed before registering a consumer; register consumers before sending data for a new id.","Treat IllegalArgumentException as a terminal state for that instruction and propagate failure to the runner rather than retrying the registration."],"exampleFix":"// before\nmultiplexer.registerConsumer(failedInstructionId, receiver);\n// after\nif (!multiplexer.isPoisoned(failedInstructionId)) {\n  multiplexer.registerConsumer(failedInstructionId, receiver);\n} else {\n  String freshId = UUID.randomUUID().toString();\n  multiplexer.registerConsumer(freshId, receiver);\n}","handlingStrategy":"validation","validationCode":"if (poisonedIds.contains(instructionId)) { useFreshInstructionId(); }","typeGuard":null,"tryCatchPattern":"try { multiplexer.registerConsumer(id, receiver); } catch (IllegalArgumentException e) { failBundle(id, e); }","preventionTips":["Never reuse instruction ids across bundles or retries","Register consumers before data flows for a new instruction","Poison only on genuine instruction failure"],"tags":["java","beam-fn","grpc-multiplexer","instruction-id"],"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"}