{"record":{"id":"1294d81acc1308b0","repo":"apache/beam","slug":"instruction-id-was-registered-twice","errorCode":null,"errorMessage":"Instruction id was registered twice","messagePattern":"Instruction id was registered twice","errorType":"exception","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java","lineNumber":122,"sourceCode":"  /**\n   * Registers a consumer for the specified instruction id.\n   *\n   * <p>The {@link BeamFnDataGrpcMultiplexer} partitions {@link BeamFnApi.Elements} with multiple\n   * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a\n   * 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.\");","sourceCodeStart":104,"sourceCodeEnd":140,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java#L104-L140","documentation":"BeamFnDataGrpcMultiplexer maps instruction IDs to inbound data receivers. Each instruction ID may be registered only once; registering the same ID again while the existing receiver is still active throws IllegalArgumentException. (Re-registering after the previous receiver completed is allowed via the complete() protocol.)","triggerScenarios":"Calling registerConsumer/registerFuture twice with the same instructionId, or a runner sending two instructions with a duplicated ID to the SDK harness.","commonSituations":"Runner-side bugs reusing instruction IDs after process bundle retries; test harness code registering both a consumer and a receiver for one ID; duplicate process-bundle requests over the BeamFnData API.","solutions":["Ensure each process bundle instruction ID is unique; regenerate the ID on retry instead of reusing it.","Call registerConsumer only once per instruction and reuse the returned receiver.","If this occurs in tests, use distinct instruction IDs or close/complete the prior receiver before re-registering."],"exampleFix":"// before\nmultiplexer.registerConsumer(instructionId, receiver);\nmultiplexer.registerConsumer(instructionId, receiver2); // duplicate\n// after\nmultiplexer.registerConsumer(instructionId, receiver);\nmultiplexer.registerConsumer(instructionId + \"-2\", receiver2);","handlingStrategy":"validation","validationCode":"Set<String> seen = new HashSet<>();\nif (!seen.add(instructionId)) throw new IllegalStateException(\"instruction id already used: \" + instructionId);","typeGuard":null,"tryCatchPattern":"try { multiplexer.registerConsumer(id, recv); } catch (IllegalArgumentException e) { /* duplicate/poisoned id: abort bundle with id context */ throw e; }","preventionTips":["Generate a fresh UUID per process-bundle instruction, including retries","Register at most one consumer/receiver per instruction ID","In runner code, assert instruction ID uniqueness before dispatch"],"tags":["beam","grpc","multiplexer","duplicate-registration"],"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-14T16:17:12.679Z"}