{"record":{"id":"2231678a43bed569","repo":"apache/beam","slug":"logging-stream-terminated-unexpectedly-before-it-was-closed","errorCode":null,"errorMessage":"Logging stream terminated unexpectedly before it was closed by the client with error: ","messagePattern":"Logging stream terminated unexpectedly before it was closed by the client with error: ","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java","lineNumber":260,"sourceCode":"\n          // Batch together as many log messages as possible that are held within the buffer\n          BeamFnApi.LogEntry.List.Builder builder =\n              BeamFnApi.LogEntry.List.newBuilder().addLogEntries(logEntry);\n          bufferedLogEntries.drainTo(additionalLogEntries);\n          builder.addAllLogEntries(additionalLogEntries);\n          outboundObserver.onNext(builder.build());\n          additionalLogEntries.clear();\n        }\n        if (inboundObserverCompletion.isDone()) {\n          try {\n            // If the inbound observer failed with an exception, get() will throw an\n            // ExecutionException.\n            inboundObserverCompletion.get();\n            // Otherwise it is an error for the server to close the stream before we closed our end.\n            throw new IllegalStateException(\n                \"Logging stream terminated unexpectedly with success before it was closed by the client.\");\n          } catch (ExecutionException e) {\n            throw new IllegalStateException(\n                \"Logging stream terminated unexpectedly before it was closed by the client with error: \"\n                    + e.getCause());\n          } catch (InterruptedException e) {\n            // Should never happen because of the isDone check.\n            Thread.currentThread().interrupt();\n            throw new RuntimeException(e);\n          }\n        }\n      } catch (OutOfMemoryError oom) {\n        throw oom;\n      } catch (Throwable t) {\n        thrown = t;\n        throw new RuntimeException(t);\n      } finally {\n        if (thrown == null) {\n          outboundObserver.onCompleted();\n        } else {\n          outboundObserver.onError(thrown);","sourceCodeStart":242,"sourceCodeEnd":278,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java#L242-L278","documentation":"BeamFnLoggingClient's drainQueueToStream waits on inboundObserverCompletion; if that future completes exceptionally, the server terminated the logging gRPC stream with an error before the client closed its end. The cause of the remote failure is appended to the message. This signals the log channel to the runner died mid-stream.","triggerScenarios":"The gRPC server-side observer of the log stream fails with an exception (ExecutionException) while the client is still sending log records — e.g. transport error, CANCELLED/UNAVAILABLE status, or server-side handler crash.","commonSituations":"Network interruption between worker and runner; worker exceeding a deadline; runner rejecting the log stream (e.g. instruction no longer known); container memory pressure killing the stream.","solutions":["Read the appended cause to identify the gRPC status and fix the underlying transport/runner failure.","Check network connectivity and timeouts between the SDK harness worker and the runner service.","Ensure the runner keeps the log stream open for the lifetime of the instruction; align Beam SDK and runner versions.","Reduce log flood that can trip flow-control or deadline limits on the stream."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  // harness/runner operation\n} catch (IllegalStateException e) {\n  Throwable cause = e.getCause();\n  if (cause instanceof StatusRuntimeException && ((StatusRuntimeException) cause).getStatus().getCode().isRetryable()) {\n    // retry or reschedule the work\n  } else { throw e; }\n}","preventionTips":["Harden network path between worker and runner","Set adequate gRPC deadlines for the log stream","Watch for runner cancellation of instructions"],"tags":["grpc","logging","network","streaming"],"backgroundTag":"broken-pipe","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}