apache/beam · error

source decode failed

Error message

source decode failed

What it means

Raised when decoding a key or parallel element from the incoming data stream fails: cp.Decode(bcr) returns an error while DataSource.invokeProcess unfolds the element. It is wrapped as 'source decode failed', meaning the coder could not interpret the byte stream for this element.

Source

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

	hasSplit := map[string]bool{}
	var checkpoints []*Checkpoint
	err := n.process(ctx, func(bcr *byteCountReader, ptransformID string) error {
		// Check if this transform has already successfully, and if so, skip reading and decoding of the elements in the buffer.
		if hasSplit[ptransformID] {
			return nil
		}
		for {
			// TODO(lostluck) 2020/02/22: Should we include window headers or just count the element sizes?
			ws, t, pn, err := DecodeWindowedValueHeader(wc, bcr.reader)
			if err != nil {
				return err
			}

			// Decode key or parallel element.
			pe, err := cp.Decode(bcr)
			if err != nil {
				return errors.Wrap(err, "source decode failed")
			}
			pe.Timestamp = t
			pe.Windows = ws
			pe.Pane = pn

			var valReStreams []ReStream
			for _, cv := range cvs {
				values, err := n.makeReStream(ctx, cv, bcr, len(cvs) == 1 && n.singleIterate)
				if err != nil {
					return err
				}
				valReStreams = append(valReStreams, values)
			}

			if err := n.Out.ProcessElement(ctx, pe, valReStreams...); err != nil {
				return err
			}
			// Collect the actual size of the element, and reset the bytecounter reader.

View on GitHub (pinned to 12126d8942)

Solutions

1. Verify the coder registered for the PTransform matches the data actually sent
2. Check for pipeline version drift between writer and reader stages; use consistent SDK versions
3. Add debugging output to the custom coder's Decode to see which bytes fail
4. Regenerate/recompile custom coders if their schema changed

Example fix

// before
func (c *myCoder) Decode(r io.Reader) (interface{}, error) {
    var b [8]byte
    io.ReadFull(r, b[:]) // length not validated
    return string(b[:]), nil
}
// after
func (c *myCoder) Decode(r io.Reader) (interface{}, error) {
    l, err := coder.DecodeVarInt(r)
    if err != nil { return nil, err }
    buf := make([]byte, l)
    if _, err := io.ReadFull(r, buf); err != nil { return nil, err }
    return string(buf), nil
}

// ensure custom coder round-trips before deploying
def roundTrip(c ElementCoder, v interface{}) bool {
    var buf bytes.Buffer
    if err := c.Encode(v, &buf); err != nil { return false }
    got, err := c.Decode(bytes.NewReader(buf.Bytes()))
    return err == nil && reflect.DeepEqual(got, v)
}

func isDecodeError(err error) bool {
    return strings.Contains(fmt.Sprint(err), "source decode failed")
}

if err != nil {
    var decErr = "source decode failed"
    if strings.Contains(err.Error(), decErr) {
        // route to dead-letter / quarantine the record
    }
}

When it happens

Trigger: A coder (cp, an ElementDecoder from the plan) tries to decode bytes read from bcr and returns an error — wrong coder, truncated payload, or malformed bytes.

Common situations: Mismatched coder versions after a pipeline update; custom coders whose Encode/Decode are not symmetric; corrupted or truncated element buffers from the runner; cross-version serialized data.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

• stream value decode failed
• CoderException
• varint too long
• %v
• unable to rewrite coder %v for state %v for transform %v in

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