{"record":{"id":"7159ce9b392e2535","repo":"apache/beam","slug":"internal-log-message-buffer-closed","errorCode":null,"errorMessage":"internal: log message buffer closed","messagePattern":"internal: log message buffer closed","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"sdks/go/pkg/beam/core/runtime/harness/logging.go","lineNumber":156,"sourceCode":"\n\tclient, err := fnpb.NewBeamFnLoggingClient(conn).Logging(ctx)\n\tif err != nil {\n\t\tconn.Close()\n\t\treturn nil, func() {}, err\n\t}\n\n\ttoDefer := func() {\n\t\tclient.CloseSend()\n\t\tconn.Close()\n\t}\n\treturn client, toDefer, nil\n}\n\ntype logSender interface {\n\tSend(*fnpb.LogEntry_List) error\n}\n\nvar errBuffClosed = errors.New(\"internal: log message buffer closed\")\n\nfunc (w *remoteWriter) connect(ctx context.Context, makeClient func(ctx context.Context) (logSender, func(), error)) error {\n\tclient, toDefer, err := makeClient(ctx)\n\tif err != nil {\n\t\treturn err\n\t}\n\tdefer toDefer()\n\n\tfor {\n\t\tconst batchSize = 64\n\t\tmsgs := make([]*fnpb.LogEntry, 0, batchSize)\n\t\tvar flush bool\n\t\tselect {\n\t\tcase <-ctx.Done():\n\t\t\treturn nil\n\t\tcase newMsg, ok := <-w.buffer:\n\t\t\tif !ok {\n\t\t\t\treturn errBuffClosed","sourceCodeStart":138,"sourceCodeEnd":174,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/logging.go#L138-L174","documentation":"The harness's remoteWriter forwards buffered log entries to the log sender over the w.buffer channel. errBuffClosed is the sentinel returned when that channel is closed while connect is still draining it — i.e. the log buffer was shut down mid-stream, an internal lifecycle condition rather than a transport failure.","triggerScenarios":"The remoteWriter's buffer channel is closed (writer shutdown/Flush completing) while connect's select loop reads from w.buffer; the ok-receive returns false and connect returns errBuffClosed.","commonSituations":"Harness shutdown racing with an in-flight log flush; logger closed before the forwarding goroutine finishes; tests exercising connect against a closed buffer.","solutions":["Treat errBuffClosed as a normal shutdown signal: compare with errors.Is and stop retrying.","Ensure Flush/shutdown ordering closes the buffer only after connect has finished.","If seen during normal operation, check for early logger Close calls in harness setup/teardown."],"exampleFix":"// before\nif err := w.connect(ctx, makeClient); err != nil { log.Fatalf(\"log connect: %v\", err) }\n// after\nif err := w.connect(ctx, makeClient); err != nil && !errors.Is(err, errBuffClosed) { log.Fatalf(\"log connect: %v\", err) }","handlingStrategy":"try-catch","validationCode":"// Only start the log forwarder when the writer is open; skip if already flushed/closed\nif w.closed { return nil }","typeGuard":"func forwardable(w *remoteWriter) bool { return w != nil && !w.closed }","tryCatchPattern":"if err := w.connect(ctx, makeClient); err != nil && !errors.Is(err, errBuffClosed) {\n    return fmt.Errorf(\"log forwarding failed: %w\", err)\n} // errBuffClosed during shutdown is expected","preventionTips":["Close the logger/buffer only after forwarding goroutines finish","Use errors.Is against errBuffClosed to treat it as normal shutdown","Test harness shutdown ordering to flush logs before teardown"],"tags":["go","apache-beam","harness","logging","channel"],"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"}