{"record":{"id":"ee34b269de05aaa6","repo":"apache/beam","slug":"failed-to-close-multiplexer","errorCode":null,"errorMessage":"Failed to close multiplexer","messagePattern":"Failed to close multiplexer","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/harness/src/main/java/org/apache/beam/fn/harness/data/BeamFnDataGrpcClient.java","lineNumber":127,"sourceCode":"      client.unregisterConsumer(instructionId);\n    }\n  }\n\n  @Override\n  public void poisonInstructionId(String instructionId) {\n    LOG.debug(\"Poisoning instruction {}\", instructionId);\n    for (BeamFnDataGrpcMultiplexer client : multiplexerCache.values()) {\n      client.poisonInstructionId(instructionId);\n    }\n  }\n\n  @Override\n  public void close() {\n    for (BeamFnDataGrpcMultiplexer client : multiplexerCache.values()) {\n      try {\n        client.close();\n      } catch (Exception e) {\n        LOG.warn(\"Failed to close multiplexer\", e);\n      }\n    }\n    multiplexerCache.clear();\n  }\n\n  @Override\n  public StreamObserver<Elements> getOutboundObserver(\n      ApiServiceDescriptor apiServiceDescriptor, String dataStreamId) {\n    return getMultiplexer(apiServiceDescriptor, dataStreamId).getOutboundObserver();\n  }\n\n  private BeamFnDataGrpcMultiplexer getMultiplexer(\n      Endpoints.ApiServiceDescriptor apiServiceDescriptor, String dataStreamId) {\n    MultiplexerKey key = new MultiplexerKey(apiServiceDescriptor, dataStreamId);\n    return multiplexerCache.computeIfAbsent(\n        key,\n        k -> {\n          OutboundObserverFactory.BasicFactory<Elements, Elements> baseOutboundObserverFactory =","sourceCodeStart":109,"sourceCodeEnd":145,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/data/BeamFnDataGrpcClient.java#L109-L145","documentation":"BeamFnDataGrpcClient.close() iterates over all cached BeamFnDataGrpcMultiplexer instances and closes them; if any multiplexer fails to close, the exception is logged with this warning and the loop continues, so one bad channel does not prevent closing the others. The cache is cleared regardless.","triggerScenarios":"A gRPC channel/multiplexer's close() throws (e.g. already-shutdown channel with pending streams, interrupted awaitTermination, transport error) while the harness data client is shutting down.","commonSituations":"Harness shutdown with in-flight data streams to the runner; job cancellation racing with gRPC calls; network problems at teardown.","solutions":["Check the logged cause; if it is an interrupted/cancelled channel it is usually benign at shutdown.","Avoid canceling the job while data streams are actively transferring; drain work first.","Inspect gRPC channel shutdown behavior and timeout configuration.","If it recurs at every startup/shutdown, update Beam version — several multiplexer close fixes landed over releases."],"exampleFix":"// before\n// nothing to change in user code; harness-level warning at shutdown\n// after\n// ensure graceful shutdown: stop producing data before closing the harness client\nstream.close();\nclient.close(); // close data streams before the multiplexer","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  multiplexer.close();\n} catch (Exception e) {\n  LOG.warn(\"Failed to close multiplexer\", e); // benign during shutdown\n}","preventionTips":["Close data streams before shutting down the harness client","Avoid cancelling jobs mid-stream when possible","Inspect the logged cause to distinguish benign shutdown races from real bugs","Keep Beam SDK up to date for multiplexer close fixes"],"tags":["java","apache-beam","grpc","shutdown"],"backgroundTag":"grpc-channel-close-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"}