{"record":{"id":"5a2278948a0c5977","repo":"apache/beam","slug":"datamgr-go-v-error-flushing-buffer-of-length-d","errorCode":null,"errorMessage":"datamgr.go [%v]: error flushing buffer of length %d","messagePattern":"datamgr\\.go \\[(.+?)\\]: error flushing buffer of length (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/harness/datamgr.go","lineNumber":641,"sourceCode":"\t\t\t\tInstructionId: string(w.id.instID),\n\t\t\t\tTransformId:   w.id.ptransformID,\n\t\t\t\tData:          w.buf,\n\t\t\t},\n\t\t},\n\t}\n\tif l := len(w.buf); l > largeBufferNotificationThreshold {\n\t\tlog.Infof(context.TODO(), \"dataWriter[%v;%v].Flush flushed large buffer of length %d\", w.id, w.ch.id, l)\n\t}\n\tw.buf = w.buf[:0]\n\treturn w.send(msg)\n}\n\nfunc (w *dataWriter) Write(p []byte) (n int, err error) {\n\tif len(w.buf)+len(p) > chunkSize {\n\t\tl := len(w.buf)\n\t\t// We can't fit this message into the buffer. We need to flush the buffer\n\t\tif err := w.Flush(); err != nil {\n\t\t\treturn 0, errors.Wrapf(err, \"datamgr.go [%v]: error flushing buffer of length %d\", w.id, l)\n\t\t}\n\t}\n\n\t// At this point there's room in the buffer one way or another.\n\tw.buf = append(w.buf, p...)\n\treturn len(p), nil\n}\n\nfunc (c *DataChannel) makeTimerWriter(ctx context.Context, id clientID, family string) *timerWriter {\n\tc.mu.Lock()\n\tdefer c.mu.Unlock()\n\treturn c.makeTimerWriterLocked(ctx, id, family)\n}\n\n// makeTimerWriterLocked does the work of makeTimerWriter, but doesn't call the lock methods.\n//\n// c.mu must be locked when this is called.\nfunc (c *DataChannel) makeTimerWriterLocked(ctx context.Context, id clientID, family string) *timerWriter {","sourceCodeStart":623,"sourceCodeEnd":659,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/harness/datamgr.go#L623-L659","documentation":"dataWriter.Write buffers bytes and flushes the existing buffer whenever an incoming write would exceed chunkSize. This error wraps a Flush failure that happened inside Write, before appending p, so nothing from p was written and 0 bytes are reported. The wrapped cause is the underlying data channel send error.","triggerScenarios":"Writing enough bytes that cumulative buffer exceeds chunkSize, triggering Flush, while the data channel's gRPC stream is dead or the send fails; large element emissions during a network partition; data service restarted mid-bundle.","commonSituations":"Long-running streaming pipelines whose channel dies silently; workers losing connection to the runner during autoscaling or preemption; emitting very large records that force frequent flushes onto a failing stream.","solutions":["Check the wrapped cause for gRPC transport errors and fix worker-to-data-service connectivity","Rely on runner-level bundle retry: since Write returns 0 bytes, the failing element can be reprocessed on retry","Add backpressure handling: pause emission when flushes fail rather than continuing to buffer","Monitor data channel health and fail the bundle promptly instead of letting writers accumulate errors"],"exampleFix":"// before\nn, err := w.Write(p)\nif err != nil { log.Fatal(err) }\n// after\nn, err := w.Write(p)\nif err != nil {\n  return fmt.Errorf(\"write to data channel failed (0 bytes consumed), failing bundle for retry: %w\", err)\n}","handlingStrategy":"try-catch","validationCode":"if len(p)+len(w.buf) > chunkSize && !w.ch.healthy() {\n  return fmt.Errorf(\"data channel %s unhealthy; large write would fail flush\", w.ch.id)\n}","typeGuard":null,"tryCatchPattern":"n, err := w.Write(p)\nif err != nil {\n  // Write consumed 0 bytes; safe to propagate for bundle retry\n  return fmt.Errorf(\"write failed at flush boundary: %w\", err)\n}","preventionTips":["Check the wrapped gRPC status to distinguish transport vs. server errors","Keep records modestly sized to reduce forced flush frequency","Ensure reliable network between workers and the data service","Rely on Write returning 0 bytes to make retries safe"],"tags":["go","apache-beam","write","flush"],"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"}