What causes "stream value decode failed" in apache/beam?

Fix the element coder so Encode/Decode round-trips all values

stream value decode failed

Error message

stream value decode failed

What it means

readStreamToBuffer decodes `size` individual elements into a buffer for a fixed-size stream; any cv.Decode failure on an element is wrapped as 'stream value decode failed'. The cause is usually a per-element decode error inside the iterable (bad bytes or an unsuitable coder).

Source

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

							r = &byteCountReader{reader: r, count: bcr.count}
							return &elementStream{r: r, ec: cv}, nil
						},
					},
				}, nil
			default:
				return nil, errors.Errorf("multi-chunk stream with invalid chunk size of %d", chunk)
			}
		}
	default:
		return nil, errors.Errorf("received stream with marker size of %d", size)
	}
}

func readStreamToBuffer(cv ElementDecoder, r io.Reader, size int64, buf []FullValue) ([]FullValue, error) {
	for i := int64(0); i < size; i++ {
		value, err := cv.Decode(r)
		if err != nil {
			return nil, errors.Wrap(err, "stream value decode failed")
		}
		buf = append(buf, *value)
	}
	return buf, nil
}

// FinishBundle resets the source.
func (n *DataSource) FinishBundle(ctx context.Context) error {
	n.mu.Lock()
	defer n.mu.Unlock()
	n.source = nil
	n.splitIdx = 0 // Ensure errors are returned for split requests if this plan is re-used.
	return n.Out.FinishBundle(ctx)
}

// Down resets the source.
func (n *DataSource) Down(ctx context.Context) error {
	n.source = nil

View on GitHub (pinned to 12126d8942)

Solutions

Fix the element coder so Encode/Decode round-trips all values
Check the stream actually contains `size` complete elements (truncation)
Test the coder in isolation with representative values
Ensure the writer and reader agree on element types

Example fix

// before
type enumCoder struct{}
func (enumCoder) Encode(v interface{}, w io.Writer) error {
    return coder.EncodeInt32(w, int32(v.(myEnum)), coder.EndOfLengthPrefix) // unvalidated
}
// after
type enumCoder struct{}
func (enumCoder) Encode(v interface{}, w io.Writer) error {
    e, ok := v.(myEnum)
    if !ok { return fmt.Errorf("expected myEnum, got %T", v) }
    return coder.EncodeInt32(w, int32(e), coder.EndOfLengthPrefix)
}

Defensive patterns

Strategy: validation

Validate before calling

// pre-check coder round-trip for element types used in GBK values
func testCoderRoundTrip(c ElementCoder, vals []interface{}) error {
    for _, v := range vals {
        var buf bytes.Buffer
        if err := c.Encode(v, &buf); err != nil { return err }
        if _, err := c.Decode(&buf); err != nil { return err }
    }
    return nil
}

Type guard

func isStreamValueDecodeFailure(err error) bool {
    return strings.Contains(err.Error(), "stream value decode failed")
}

Try / catch

if err != nil && strings.Contains(err.Error(), "stream value decode failed") {
    // quarantine element; check for truncation (EOF) vs. encoding error
}

Prevention

Round-trip test coders over representative data
Ensure the stream carries exactly `size` complete elements
Handle all enum/range values in custom coders

When it happens

Trigger: While reading N elements of a sized stream, element i's cv.Decode(r) returns an error (EOF mid-element, invalid encoding, custom coder failure).

Common situations: GBK results with elements encoded by a mismatched coder; truncated stream ending before all N elements are read; custom coders failing on particular values (e.g. invalid UTF-8, out-of-range enums).

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

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