What causes "datamgr.go [%v]: error flushing buffer of length %d" in apache/beam?

Check the wrapped cause for gRPC transport errors and fix worker-to-data-service connectivity

datamgr.go [ ]: error flushing buffer of length

Error message

datamgr.go [%v]: error flushing buffer of length %d

What it means

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.

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

Example fix

// before
n, err := w.Write(p)
if err != nil { log.Fatal(err) }
// after
n, err := w.Write(p)
if err != nil {
  return fmt.Errorf("write to data channel failed (0 bytes consumed), failing bundle for retry: %w", err)
}

Defensive patterns

Strategy: try-catch

Validate before calling

if len(p)+len(w.buf) > chunkSize && !w.ch.healthy() {
  return fmt.Errorf("data channel %s unhealthy; large write would fail flush", w.ch.id)
}

Try / catch

n, err := w.Write(p)
if err != nil {
  // Write consumed 0 bytes; safe to propagate for bundle retry
  return fmt.Errorf("write failed at flush boundary: %w", err)
}

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/5a2278948a0c5977. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/harness/datamgr.go:641

				InstructionId: string(w.id.instID),
				TransformId:   w.id.ptransformID,
				Data:          w.buf,
			},
		},
	}
	if l := len(w.buf); l > largeBufferNotificationThreshold {
		log.Infof(context.TODO(), "dataWriter[%v;%v].Flush flushed large buffer of length %d", w.id, w.ch.id, l)
	}
	w.buf = w.buf[:0]
	return w.send(msg)
}

func (w *dataWriter) Write(p []byte) (n int, err error) {
	if len(w.buf)+len(p) > chunkSize {
		l := len(w.buf)
		// We can't fit this message into the buffer. We need to flush the buffer
		if err := w.Flush(); err != nil {
			return 0, errors.Wrapf(err, "datamgr.go [%v]: error flushing buffer of length %d", w.id, l)
		}
	}

	// At this point there's room in the buffer one way or another.
	w.buf = append(w.buf, p...)
	return len(p), nil
}

func (c *DataChannel) makeTimerWriter(ctx context.Context, id clientID, family string) *timerWriter {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.makeTimerWriterLocked(ctx, id, family)
}

// makeTimerWriterLocked does the work of makeTimerWriter, but doesn't call the lock methods.
//
// c.mu must be locked when this is called.
func (c *DataChannel) makeTimerWriterLocked(ctx context.Context, id clientID, family string) *timerWriter {

View on GitHub (pinned to 12126d8942)