{"record":{"id":"cb3eeaa1ddc4d9d5","repo":"apache/beam","slug":"unexpected-error-closing-existing-observer","errorCode":null,"errorMessage":"Unexpected error closing existing observer","messagePattern":"Unexpected error closing existing observer","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java","lineNumber":162,"sourceCode":"  /**\n   * Poisons an instruction id.\n   *\n   * <p>Any records for the instruction on the inbound observer will be dropped for the next {@link\n   * #POISONED_INSTRUCTION_ID_CACHE_TIMEOUT}.\n   */\n  public void poisonInstructionId(String instructionId) {\n    poisonedInstructionIds.put(instructionId, Boolean.TRUE);\n    @Nullable CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiverFuture =\n        receivers.remove(instructionId);\n    if (receiverFuture != null) {\n      // Completing exceptionally has no effect if the future was already notified. In that case\n      // whatever registered the receiver needs to handle cancelling it.\n      receiverFuture.completeExceptionally(new PoisonedException());\n      if (!receiverFuture.isCompletedExceptionally()) {\n        try {\n          receiverFuture.get().close();\n        } catch (Exception e) {\n          LOG.warn(\"Unexpected error closing existing observer\");\n        }\n      }\n    }\n  }\n\n  @VisibleForTesting\n  boolean hasConsumer(String instructionId) {\n    return receivers.containsKey(instructionId);\n  }\n\n  @Override\n  public void close() throws Exception {\n    Exception exception = null;\n    for (CompletableFuture<CloseableFnDataReceiver<BeamFnApi.Elements>> receiver :\n        ImmutableList.copyOf(receivers.values())) {\n      // Cancel any observer waiting for the client to complete. If the receiver has already been\n      // completed or cancelled, this call will be ignored.\n      receiver.cancel(true);","sourceCodeStart":144,"sourceCodeEnd":180,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/fn/data/BeamFnDataGrpcMultiplexer.java#L144-L180","documentation":"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).","triggerScenarios":"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.","commonSituations":"Races between instruction poisoning and the SDK/harness finishing the instruction normally; closed gRPC channels during shutdown; duplicate instruction ids.","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."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// This is an internal WARN log, not a thrown exception; guard against its consequences:\ntry {\n  multiplexer.poisonInstructionId(instructionId);\n} catch (RuntimeException e) {\n  LOG.warn(\"Poisoning failed for instruction {}\", instructionId, e);\n}","preventionTips":["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."],"tags":["grpc","streaming","beam","cleanup"],"backgroundTag":"resource-cleanup-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}