apache/beam · error

error decoding []byte field

Error message

error decoding []byte field

What it means

This error wraps a failure from DecodeBytes while reflectively decoding a []byte field in a Beam row coder. The underlying bytes decoding (length prefix or truncated stream) failed, and the wrapper adds context naming the field kind.

Solutions

1. Inspect the wrapped cause (errors.Unwrap) to see if it is EOF/unexpected-EOF and fix the producer to write complete length-prefixed byte slices
2. Ensure both pipeline stages use the same Beam SDK version so byte-slice encodings match
3. Re-run the job; if data corruption in the transport is suspected, verify serialization between stages

if len(data) == 0 { return fmt.Errorf("empty byte field") }

if _, err := decodeRow(r); err != nil {
    cause := err
    for errors.Unwrap(cause) != nil { cause = errors.Unwrap(cause) }
    log.Printf("byte field decode failed: %v", cause)
    return err
}

When it happens

Trigger: Decoding a Beam pipeline element whose schema has a []byte field, when the byte-slice payload in the stream is truncated, corrupt, or was encoded by an incompatible coder version.

Common situations: Mismatched pipeline SDK versions between writer and runner, corrupted/short data in a test transport, or a custom coder writing rows without the expected length prefix.

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

• bad combinefn
• bad decoding function
• bad encoding function
• bad function
• bad function type

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

Appendix: source

Thrown at sdks/go/pkg/beam/core/graph/coder/row_decoder.go:265

		return errors.Wrap(err, "error decoding single-precision float field")
	}
	rv.SetFloat(float64(v))
	return nil
}

func reflectDecodeFloat(rv reflect.Value, r io.Reader) error {
	v, err := DecodeDouble(r)
	if err != nil {
		return errors.Wrap(err, "error decoding double field")
	}
	rv.SetFloat(v)
	return nil
}

func reflectDecodeByteSlice(rv reflect.Value, r io.Reader) error {
	b, err := DecodeBytes(r)
	if err != nil {
		return errors.Wrap(err, "error decoding []byte field")
	}
	rv.SetBytes(b)
	return nil
}

// customFunc returns nil if no custom func exists for this type.
// If an error is returned, coder construction should be aborted.
func (b *RowDecoderBuilder) customFunc(t reflect.Type) (func(io.Reader) (any, error), bool, error) {
	if fact, ok := b.allFuncs[t]; ok {
		f, err := fact(t)

		if err != nil {
			return nil, false, err
		}
		return f, false, nil
	}
	// Check satisfaction of interface types in reverse registration order.
	pt := reflect.PtrTo(t)

View on GitHub (pinned to 12126d8942)