{"record":{"id":"0bafa5f8397b0278","repo":"apache/beam","slug":"inboundobserver-for-beamfnstatusclient-completed-with","errorCode":null,"errorMessage":"InboundObserver for BeamFnStatusClient completed with exception.","messagePattern":"InboundObserver for BeamFnStatusClient completed with exception\\.","errorType":"console","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/BeamFnStatusClient.java","lineNumber":92,"sourceCode":"    this.inboundObserverCompletion = new CompletableFuture<>();\n    Thread thread = new Thread(memoryMonitor);\n    thread.setDaemon(true);\n    thread.setPriority(Thread.MIN_PRIORITY);\n    thread.setName(\"MemoryMonitor\");\n    thread.start();\n\n    // Start the rpc after all the initialization is complete as the InboundObserver\n    // may be called any time after this.\n    this.outboundObserver =\n        BeamFnWorkerStatusGrpc.newStub(channel).workerStatus(new InboundObserver());\n  }\n\n  @Override\n  public void close() throws Exception {\n    try {\n      Object completion = inboundObserverCompletion.get(1, TimeUnit.MINUTES);\n      if (completion != COMPLETED) {\n        LOG.warn(\"InboundObserver for BeamFnStatusClient completed with exception.\");\n      }\n    } finally {\n      // Shut the channel down\n      channel.shutdown();\n      if (!channel.awaitTermination(10, TimeUnit.SECONDS)) {\n        channel.shutdownNow();\n      }\n    }\n  }\n\n  /**\n   * Class representing the execution state of a thread.\n   *\n   * <p>Can be used in hash maps.\n   */\n  static class Stack {\n    final StackTraceElement[] elements;\n    final Thread.State state;","sourceCodeStart":74,"sourceCodeEnd":110,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/status/BeamFnStatusClient.java#L74-L110","documentation":"BeamFnStatusClient.close() waits up to 1 minute for the worker status inbound observer (used for worker status/harness health reporting) to complete; if the completion latch is not COMPLETED after the wait, this warning is logged before shutting the channel down. It means the status stream ended abnormally (exception, timeout, or never finished) but close() proceeds regardless.","triggerScenarios":"Closing the status client while the inbound observer (WorkerStatusListener stream) has not completed within 1 minute — e.g. runner never closed the status stream, the stream errored, or the thread waiting timed out.","commonSituations":"Harness shutdown while the runner keeps the status channel open; job cancellation not closing the status stream; network partition leaving the status stream half-open.","solutions":["Check whether the status stream errored earlier in the logs (grpc error before shutdown).","Ensure the runner closes the worker status stream during job teardown.","If benign at process exit, ignore — channel.shutdown() still proceeds.","Upgrade Beam if this occurs on every clean shutdown; stream completion handling has been fixed in later versions."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"Object completion = inboundObserverCompletion.get(1, TimeUnit.MINUTES);\nif (completion != COMPLETED) {\n  LOG.warn(\"InboundObserver for BeamFnStatusClient completed with exception.\");\n}\n// finally: channel.shutdown(); awaitTermination(10s); shutdownNow() if needed;","preventionTips":["Ensure the runner closes the worker status stream at job teardown","Look for earlier gRPC errors on the status channel","Expect benign occurrences at abrupt process exit","Upgrade Beam if it occurs on every clean shutdown"],"tags":["java","apache-beam","grpc","status-client","shutdown"],"backgroundTag":"grpc-stream-not-completed","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"}