What causes "Unexpected error closing existing observer" in apache/beam?

Generally safe to ignore — it is a logged warning during cleanup; check for a larger pattern of poisoned instructions.

Unexpected error closing existing observer

Error message

Unexpected error closing existing observer

What it means

When BeamFnDataGrpcMultiplexer poisons an instruction id, any outstanding inbound receiver future is completed exceptionally; if it turns out the receiver was still completed normally, the code closes the observer and logs any failure to do so. This log line indicates a non-fatal cleanup problem — the observer's close() threw (e.g. because the stream was already closed/cancelled).

Solutions

Generally safe to ignore — it is a logged warning during cleanup; check for a larger pattern of poisoned instructions.
If frequent, investigate why instructions are being poisoned (look for the original poisoning cause logged elsewhere).
Verify the gRPC channel/connection is not being closed prematurely by the harness.
Check for duplicate or reused instruction ids in the runner.

Defensive patterns

Strategy: try-catch

Try / catch

// This is an internal WARN log, not a thrown exception; guard against its consequences:
try {
  multiplexer.poisonInstructionId(instructionId);
} catch (RuntimeException e) {
  LOG.warn("Poisoning failed for instruction {}", instructionId, e);
}

Prevention

Monitor logs for frequent poisoning — it usually signals a deeper harness/runner issue.
Avoid reusing gRPC instruction ids.
Ensure the FnConnection channel is closed only after all inflight instructions complete.

When it happens

Trigger: poisonInstructionId is called for an instruction whose inbound receiver future completed successfully (e.g. via a late inbound response), and receiverFuture.get().close() throws an IOException from the underlying gRPC stream.

Common situations: Races between instruction poisoning and the SDK/harness finishing the instruction normally; closed gRPC channels during shutdown; duplicate instruction ids.

Related errors

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

Appendix: source

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

  /**
   * Poisons an instruction id.
   *
   * <p>Any records for the instruction on the inbound observer will be dropped for the next {@link
   * #POISONED_INSTRUCTION_ID_CACHE_TIMEOUT}.
   */
  public void poisonInstructionId(String instructionId) {
    poisonedInstructionIds.put(instructionId, Boolean.TRUE);
    @Nullable CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture =
        receivers.remove(instructionId);
    if (receiverFuture != null) {
      // Completing exceptionally has no effect if the future was already notified. In that case
      // whatever registered the receiver needs to handle cancelling it.
      receiverFuture.completeExceptionally(new PoisonedException());
      if (!receiverFuture.isCompletedExceptionally()) {
        try {
          receiverFuture.get().close();
        } catch (Exception e) {
          LOG.warn("Unexpected error closing existing observer");
        }
      }
    }
  }

  @VisibleForTesting
  boolean hasConsumer(String instructionId) {
    return receivers.containsKey(instructionId);
  }

  @Override
  public void close() throws Exception {
    Exception exception = null;
    for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :
        ImmutableList.copyOf(receivers.values())) {
      // Cancel any observer waiting for the client to complete. If the receiver has already been
      // completed or cancelled, this call will be ignored.
      receiver.cancel(true);

View on GitHub (pinned to 12126d8942)