apache/beam · error · IllegalArgumentException

Instruction id was registered twice

Error message

Instruction id was registered twice

What it means

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.)

Source

Thrown at sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java:122

  /**
   * Registers a consumer for the specified instruction id.
   *
   * <p>The {@link BeamFnDataGrpcMultiplexer} partitions {@link BeamFnApi.Elements} with multiple
   * instruction ids ensuring that the receiver will only see {@link BeamFnApi.Elements} with a
   * single instruction id.
   *
   * <p>The caller must either {@link #unregisterConsumer unregister the consumer} when all messages
   * have been processed or {@link #poisonInstructionId(String) poison the instruction} if messages
   * for the instruction should be dropped.
   */
  public void registerConsumer(
      String instructionId, CloseableFnDataReceiver<BeamFnApi.Elements> receiver) {
    receivers.compute(
        instructionId,
        (unused, existing) -> {
          if (existing != null) {
            if (!existing.complete(receiver)) {
              throw new IllegalArgumentException("Instruction id was registered twice");
            }
            return existing;
          }
          if (poisonedInstructionIds.getIfPresent(instructionId) != null) {
            throw new IllegalArgumentException("Instruction id was poisoned");
          }
          return CompletableFuture.completedFuture(receiver);
        });
  }

  /** Unregisters a previously registered consumer. */
  public void unregisterConsumer(String instructionId) {
    @Nullable CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture =
        receivers.remove(instructionId);
    if (receiverFuture != null && !receiverFuture.isDone()) {
      // The future must have been inserted by the inbound observer since registerConsumer completes
      // the future.
      throw new IllegalArgumentException("Unregistering consumer which was not registered.");

View on GitHub (pinned to 12126d8942)

Solutions

1. Ensure each process bundle instruction ID is unique; regenerate the ID on retry instead of reusing it.
2. Call registerConsumer only once per instruction and reuse the returned receiver.
3. If this occurs in tests, use distinct instruction IDs or close/complete the prior receiver before re-registering.

Example fix

// before
multiplexer.registerConsumer(instructionId, receiver);
multiplexer.registerConsumer(instructionId, receiver2); // duplicate
// after
multiplexer.registerConsumer(instructionId, receiver);
multiplexer.registerConsumer(instructionId + "-2", receiver2);

Set<String> seen = new HashSet<>();
if (!seen.add(instructionId)) throw new IllegalStateException("instruction id already used: " + instructionId);

try { multiplexer.registerConsumer(id, recv); } catch (IllegalArgumentException e) { /* duplicate/poisoned id: abort bundle with id context */ throw e; }

When it happens

Trigger: Calling registerConsumer/registerFuture twice with the same instructionId, or a runner sending two instructions with a duplicated ID to the SDK harness.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• State stream is closed.
• Unknown type
• Outbound data endpoint already registered for
• Outbound timers endpoint already registered for ${timerKey}
• Conflicting registrations for: %s

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/1294d81acc1308b0. Report an issue: GitHub.