apache/beam · error

stream size decoding failed

Error message

stream size decoding failed

What it means

In makeReStream, the size marker of a nested (re-iterable) stream is read with coder.DecodeInt32. If that read fails (EOF, corrupt bytes), the error is wrapped as 'stream size decoding failed'. This is a wire-format failure while unpacking GBK/CoGBK result iterables.

Solutions

1. Check the upstream writer's coder matches the reader's expectation
2. Look for earlier decode errors in the log that would explain stream misalignment
3. Re-run the failing bundle; transient runner data issues may resolve
4. Verify no custom coder writes a different stream framing than the standard size-prefixed format

// stream framing errors are not retryable in user code; rely on runner retry
if strings.Contains(err.Error(), "stream size decoding failed") {
    log.Errorf("corrupt stream framing: %v", err)
    return err
}

When it happens

Trigger: makeReStream is called for a GBK/CoGBK value stream; coder.DecodeInt32(bcr.reader) hits EOF or an invalid varint because the stream is truncated or misaligned.

Common situations: Corrupted shuffle data; coder misalignment where an upstream stage writes a different format than expected; short reads when a bundle's data buffer ends mid-stream.

Related errors

• stream chunk size decoding failed
• decodeMultiChunkStream chunk size decoding failed
• decodeStream value decode failed on close
• decoding a *
• decoding a

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

Appendix: source

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

			if node, ok := n.OnTimerTransforms[ptransformID]; ok {
				if err := node.ProcessTimers(timerFamilyID, bcr); err != nil {
					log.Warnf(ctx, "expected transform %v to have an OnTimer method attached to handle"+
						"Timer Family ID: %v callback, but it did not. Please file an issue with Apache Beam"+
						"if you have defined OnTimer method with reproducible code at https://github.com/apache/beam/issues", ptransformID, timerFamilyID)
					return errors.WithContext(err, "ontimer callback invocation failed")
				}
			}
			return nil
		})

	return checkpoints, err
}

func (n *DataSource) makeReStream(ctx context.Context, cv ElementDecoder, bcr *byteCountReader, onlyStream bool) (ReStream, error) {
	// TODO(lostluck) 2020/02/22: Do we include the chunk size, or just the element sizes?
	size, err := coder.DecodeInt32(bcr.reader)
	if err != nil {
		return nil, errors.Wrap(err, "stream size decoding failed")
	}

	if onlyStream {
		// If we know the stream won't be re-iterated,
		// decode elements on demand instead to reduce memory usage.
		switch {
		case size >= 0:
			return &singleUseReStream{
				r:    bcr,
				d:    cv,
				size: int(size),
			}, nil
		case size == -1:
			return &singleUseMultiChunkReStream{
				r: bcr,
				d: cv,
				open: func(bcr *byteCountReader) (Stream, error) {
					tokenLen, err := coder.DecodeVarInt(bcr.reader)

View on GitHub (pinned to 12126d8942)