dataWriter[ ; ].Close: error flushing buffer of length

Error message

dataWriter[%v;%v].Close: error flushing buffer of length %d

What it means

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.

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

Example fix

// before
if err := w.Close(); err != nil { log.Printf("drop: %v", err) }
// after
if err := w.Close(); err != nil {
  return fmt.Errorf("final flush failed, elements may be lost: %w", err)
}

Defensive patterns

Strategy: try-catch

Validate before calling

if !dataChannelHealthy(w.ch) {
  return fmt.Errorf("data channel %s unhealthy before close; flush will fail", w.ch.id)
}

Try / catch

if err := w.Close(); err != nil {
  var cause error
  errors.As(err, &cause)
  if isTransportError(cause) {
    return fmt.Errorf("elements may be lost, bundle should be retried: %w", err)
  }
  return err
}

Prevention

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

When it happens

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

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

Related errors

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

Appendix: source

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

				// Per GRPC stream documentation, if there's an EOF, we must call Recv
				// until a non-nil error is returned, to ensure resources are cleaned up.
				// https://pkg.go.dev/google.golang.org/grpc#ClientConn.NewStream
				_, err = w.ch.client.Recv()
			}
		}
		log.Warnf(context.TODO(), "dataWriter[%v;%v] error on send: %v", w.id, w.ch.id, err)
		w.ch.terminateStreamOnError(err)
		return err
	}
	return nil
}

func (w *dataWriter) Close() error {
	// Don't acquire the locks as Flush will do so.
	l := len(w.buf)
	err := w.Flush()
	if err != nil {
		return errors.Wrapf(err, "dataWriter[%v;%v].Close: error flushing buffer of length %d", w.id, w.ch.id, l)
	}
	// TODO(BEAM-13082): Consider a sync.Pool to reuse < 64MB buffers.
	// The dataWriter won't be reused, but may be referenced elsewhere.
	// Drop the buffer to let it be GC'd.
	w.buf = nil

	// Now acquire the locks since we're sending.
	w.ch.mu.Lock()
	defer w.ch.mu.Unlock()
	delete(w.ch.writers[w.id.instID], w.id.ptransformID)
	msg := &fnpb.Elements{
		Data: []*fnpb.Elements_Data{
			{
				InstructionId: string(w.id.instID),
				TransformId:   w.id.ptransformID,
				// TODO(https://github.com/apache/beam/issues/21164): Set IsLast true on final flush instead of w/empty sentinel?
				// Empty data == sentinel
				IsLast: true,

View on GitHub (pinned to 12126d8942)