{"record":{"id":"817c5af6f3768724","repo":"apache/beam","slug":"logging-stream-terminated-unexpectedly-with-success-before","errorCode":null,"errorMessage":"Logging stream terminated unexpectedly with success before it was closed by the client.","messagePattern":"Logging stream terminated unexpectedly with success before it was closed by the client\\.","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java","lineNumber":257,"sourceCode":"            }\n            continue;\n          }\n\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) {","sourceCodeStart":239,"sourceCodeEnd":275,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/logging/BeamFnLoggingClient.java#L239-L275","documentation":"BeamFnLoggingClient streams log messages from the SDK harness to the runner over a gRPC stream. This IllegalStateException is thrown by drainQueueToStream when the server half of the logging stream completes with success before the client closed its own end of the stream, which the Fn API logging protocol forbids. It indicates the runner terminated the log stream early, so subsequent log messages could not be delivered.","triggerScenarios":"The runner/server calls onCompleted() on the inbound logging observer while the client is still draining its log queue to the stream (client has not yet called close on its request observer).","commonSituations":"Runner shutting down a bundle or worker while the SDK harness still logs; runner bugs or aggressive timeouts closing the instruction logging stream early; pipeline finishing while deferred log messages remain queued.","solutions":["Check the runner version/health — this usually indicates the runner closed the logging stream prematurely; upgrade runner and Beam SDK to matching versions.","Inspect for errors/aborts around the same instruction that would cause the runner to end the stream early (bundle failure, cancellation).","Reduce excessive logging volume that keeps the queue draining long after the runner considers the stage done.","Retry/re-run the pipeline; if reproducible, file an issue with runner logs."],"exampleFix":"// before: relying on logging during/after stream close\nLOG.info(\"stage finished after cleanup\");\n// after: guard logging so it stops once the stream is closing\nif (!loggingClient.isClosed()) {\n  LOG.info(\"stage finished after cleanup\");\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  // pipeline work with logging\n} catch (IllegalStateException e) {\n  if (e.getMessage().contains(\"terminated unexpectedly\")) {\n    LOG.warn(\"Log stream closed early by runner; logs may be incomplete\", e);\n  } else { throw e; }\n}","preventionTips":["Keep Beam SDK and runner versions aligned","Avoid logging after pipeline teardown begins","Monitor runner-side stream cancellation causes"],"tags":["grpc","logging","streaming","beam"],"backgroundTag":"unexpected-api-response-shape","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"}