apache/beam · error

source failed processing data

Error message

source failed processing data

What it means

The DataSource process() loop wraps any error returned by the user DoFn's process callback while consuming a data element. The underlying error is preserved via errors.Wrapf; io.EOF is excluded because it signals a successfully drained reader, not a failure. This means the actual failure is a user DoFn error or a nested decode/process error, surfaced at the source boundary.

Solutions

1. Inspect the wrapped (cause) error in the message chain to find the failing DoFn or decoder
2. Fix the user DoFn code that returned the error for the offending record
3. Re-verify coder consistency between pipeline construction and runtime (same SDK versions)
4. If caused by a specific element, filter or handle bad records defensively in the DoFn

Example fix

// before
func (fn *myFn) ProcessElement(v string) error {
    return doSomething(v) // may fail on bad input
}
// after
func (fn *myFn) ProcessElement(v string) error {
    if err := doSomething(v); err != nil {
        // handle or fall back instead of failing the bundle
        return handleGracefully(v, err)
    }
    return nil
}

// DoFn process errors propagate as wrapped 'source failed processing data'
if err := dofn.ProcessElement(v); err != nil {
    // log and decide: fail bundle vs. skip record
    log.Errorf("element %v failed: %v", v, err)
    return err // or return nil to skip
}

When it happens

Trigger: A data element arrives on the runner's data channel with len(e.Data)>0; r.Reset(e.Data) succeeds but the `data(&bcr, e.PtransformID)` callback (which decodes elements and invokes the DoFn) returns a non-nil error other than io.EOF.

Common situations: User DoFn panics/returns errors inside processElement; custom coders failing to decode; downstream transform raising errors on specific records; corrupted element payloads from an incompatible pipeline version.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• bad coder kind
• bad CoGBK
• bad combinefn
• bad decode
• bad decoding function

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/runtime/exec/datasource.go:146

	var byteCount int
	bcr := byteCountReader{reader: &r, count: &byteCount}

	for {
		n.consumingReceivedData.Store(false)
		var err error
		select {
		case e, ok := <-elms:
			n.consumingReceivedData.Store(true)
			// Channel closed, so time to exit
			if !ok {
				return nil
			}
			if len(e.Data) > 0 {
				r.Reset(e.Data)
				err = data(&bcr, e.PtransformID)
			}
			if err != nil && err != io.EOF {
				return errors.Wrapf(err, "source failed processing data")
			}
			// Process any simultaneously sent timers.
			// If the data channel has split though
			if len(e.Timers) > 0 {
				r.Reset(e.Timers)
				err = timer(&bcr, e.PtransformID, e.TimerFamilyID)
			}
			if err != nil && err != io.EOF {
				return errors.Wrap(err, "source failed processing timers")
			}
			// io.EOF means the reader successfully drained.
			// We're ready for a new buffer.
		case <-ctx.Done():
			// now that it is done processing received data, we set it to false.
			n.consumingReceivedData.Store(false)
			return nil
		}
	}

View on GitHub (pinned to 12126d8942)