apache/beam · error

decodeStream value decode failed on close

Error message

decodeStream value decode failed on close

What it means

During Close of a length-prefixed decode stream, any remaining undecoded values within the declared size are drained and decoded; a decode failure there is wrapped as 'decodeStream value decode failed on close'. This means the serialized bytes on the stream are not decodable with the expected coder, surfacing at stream shutdown rather than during normal reads.

Source

Thrown at sdks/go/pkg/beam/core/runtime/exec/fullvalue.go:237

	r          io.Reader
	d          ElementDecoder
	next, size int
	ret        FullValue
}

// Close causes subsequent calls to Read to return io.EOF, and drains the remaining element count
// from the reader.
func (s *decodeStream) Close() error {
	// On close, if next != size, we must iterate through the rest of the decoding
	// until the reader is drained. Otherwise we corrupt the read for the next element.
	//
	// TODO(https://github.com/apache/beam/issues/22901):
	// Optimize the case where we have length prefixed values
	// so we can avoid allocating the values in the first place.
	for s.next < s.size {
		err := s.d.DecodeTo(s.r, &s.ret)
		if err != nil {
			return errors.Wrap(err, "decodeStream value decode failed on close")
		}
		s.next++
	}
	s.r = nil
	s.d = nil
	s.ret = FullValue{}
	return nil
}

// Read produces the next value in the stream.
func (s *decodeStream) Read() (*FullValue, error) {
	if s.r == nil || s.next == s.size {
		return nil, io.EOF
	}
	err := s.d.DecodeTo(s.r, &s.ret)
	if err != nil {
		if err == io.EOF {
			return nil, io.EOF

View on GitHub (pinned to 12126d8942)

Solutions

1. Retry the bundle/stage; transient data-plane corruption usually resolves on rerun.
2. Verify all workers run the same SDK version as the pipeline was expanded with (coder compatibility).
3. Inspect custom coders/encoders registered for the element type for malformed output.
4. Check runner logs for transport errors (broken pipe, cancelled gRPC stream) preceding this failure.

// runner-side: check the wrapped cause
if strings.Contains(err.Error(), "decodeStream value decode failed on close") {
    log.Warnf("data stream corruption at close; retrying bundle: %v", errors.Unwrap(err))
    return retryable // surface as retryable to the runner
}

When it happens

Trigger: Close() on the decode stream while s.next < s.size and s.d.DecodeTo(s.r, &s.ret) fails: truncated/corrupted runner-side data stream, coder mismatch between writer and reader, or a partially delivered gRPC element stream that ends mid-record.

Common situations: Runner/data-plane instability dropping bytes mid-stream; mismatched coder versions between SDK workers (e.g. pipeline built with a coder the executing worker doesn't support); custom coder emitting malformed bytes.

Related errors

• decodeMultiChunkStream chunk size decoding failed
• error decoding bool: %v
• invalid float encoding for: %v
• invalid varintz encoding for: %v
• invalid varuintz encoding for: %v

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