{"record":{"id":"9716954549f6e4ee","repo":"apache/beam","slug":"datawriter-v-v-close-error-flushing-buffer-of-length-d","errorCode":null,"errorMessage":"dataWriter[%v;%v].Close: error flushing buffer of length %d","messagePattern":"dataWriter\\[(.+?);(.+?)\\]\\.Close: error flushing buffer of length (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/harness/datamgr.go","lineNumber":586,"sourceCode":"\t\t\t\t// Per GRPC stream documentation, if there's an EOF, we must call Recv\n\t\t\t\t// until a non-nil error is returned, to ensure resources are cleaned up.\n\t\t\t\t// https://pkg.go.dev/google.golang.org/grpc#ClientConn.NewStream\n\t\t\t\t_, err = w.ch.client.Recv()\n\t\t\t}\n\t\t}\n\t\tlog.Warnf(context.TODO(), \"dataWriter[%v;%v] error on send: %v\", w.id, w.ch.id, err)\n\t\tw.ch.terminateStreamOnError(err)\n\t\treturn err\n\t}\n\treturn nil\n}\n\nfunc (w *dataWriter) Close() error {\n\t// Don't acquire the locks as Flush will do so.\n\tl := len(w.buf)\n\terr := w.Flush()\n\tif err != nil {\n\t\treturn errors.Wrapf(err, \"dataWriter[%v;%v].Close: error flushing buffer of length %d\", w.id, w.ch.id, l)\n\t}\n\t// TODO(BEAM-13082): Consider a sync.Pool to reuse < 64MB buffers.\n\t// The dataWriter won't be reused, but may be referenced elsewhere.\n\t// Drop the buffer to let it be GC'd.\n\tw.buf = nil\n\n\t// Now acquire the locks since we're sending.\n\tw.ch.mu.Lock()\n\tdefer w.ch.mu.Unlock()\n\tdelete(w.ch.writers[w.id.instID], w.id.ptransformID)\n\tmsg := &fnpb.Elements{\n\t\tData: []*fnpb.Elements_Data{\n\t\t\t{\n\t\t\t\tInstructionId: string(w.id.instID),\n\t\t\t\tTransformId:   w.id.ptransformID,\n\t\t\t\t// TODO(https://github.com/apache/beam/issues/21164): Set IsLast true on final flush instead of w/empty sentinel?\n\t\t\t\t// Empty data == sentinel\n\t\t\t\tIsLast: true,","sourceCodeStart":568,"sourceCodeEnd":604,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/datamgr.go#L568-L604","documentation":"dataWriter.Close flushes any buffered bytes to the underlying data channel; this error wraps a Flush failure that occurred during Close. The writer had l bytes still buffered that could not be delivered to the data service, so those bytes are lost for this write path. It typically reflects an already-broken gRPC stream underneath.","triggerScenarios":"Calling Close on a dataWriter whose buffer exceeds 0 and whose Flush fails because the DataChannel's gRPC stream errored or was closed; the data service went away mid-bundle; the channel was closed concurrently.","commonSituations":"Network interruption between worker and runner while elements are buffered; runner canceling the bundle while a DoFn is still emitting and then closing writers; oversized buffered chunks timing out on send.","solutions":["Inspect the wrapped cause: if it is a gRPC Unavailable/transport error, fix connectivity between worker and data service and rerun the bundle","Ensure the bundle is not canceled while DoFns are still emitting; complete or abort writers before teardown","Reduce buffered volume per channel or flush more frequently to shrink exposure to stream failures","Retry the pipeline/bundle; buffered element loss on a failed flush makes retry the correct recovery"],"exampleFix":"// before\nif err := w.Close(); err != nil { log.Printf(\"drop: %v\", err) }\n// after\nif err := w.Close(); err != nil {\n  return fmt.Errorf(\"final flush failed, elements may be lost: %w\", err)\n}","handlingStrategy":"try-catch","validationCode":"if !dataChannelHealthy(w.ch) {\n  return fmt.Errorf(\"data channel %s unhealthy before close; flush will fail\", w.ch.id)\n}","typeGuard":null,"tryCatchPattern":"if err := w.Close(); err != nil {\n  var cause error\n  errors.As(err, &cause)\n  if isTransportError(cause) {\n    return fmt.Errorf(\"elements may be lost, bundle should be retried: %w\", err)\n  }\n  return err\n}","preventionTips":["Flush proactively in long-running DoFns so Close has little buffered data","Treat Close errors on writers as bundle failures, never log-and-continue","Monitor gRPC stream state and fail fast on breakage","Avoid canceling bundles while emitters are active"],"tags":["go","apache-beam","flush","data-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-20T03:17:13.778Z"}