apache/beam · error · IllegalArgumentException

Instruction id was poisoned

Error message

Instruction id was poisoned

What it means

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.

Source

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

   * 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.");
    }
  }

  /**
   * Poisons an instruction id.

View on GitHub (pinned to 12126d8942)

Solutions

1. Use a fresh, unique instruction id when retrying after failure instead of reusing the poisoned one.
2. Check whether the instruction failed before registering a consumer; register consumers before sending data for a new id.
3. Treat IllegalArgumentException as a terminal state for that instruction and propagate failure to the runner rather than retrying the registration.

Example fix

// before
multiplexer.registerConsumer(failedInstructionId, receiver);
// after
if (!multiplexer.isPoisoned(failedInstructionId)) {
  multiplexer.registerConsumer(failedInstructionId, receiver);
} else {
  String freshId = UUID.randomUUID().toString();
  multiplexer.registerConsumer(freshId, receiver);
}

if (poisonedIds.contains(instructionId)) { useFreshInstructionId(); }

try { multiplexer.registerConsumer(id, receiver); } catch (IllegalArgumentException e) { failBundle(id, e); }

When it happens

Trigger: Calling registerConsumer(instructionId, receiver) after the instruction was poisoned via poison(instructionId), typically following a processing failure on the instruction.

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

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

• Unregistering consumer which was not registered.
• PoisonedException
• UnsupportedOperationException
• NoSuchElementException
• Unable to find inbound data receiver for instruction %s and

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