{"record":{"id":"df58a0af3367ca2b","repo":"apache/beam","slug":"stream-size-decoding-failed","errorCode":null,"errorMessage":"stream size decoding failed","messagePattern":"stream size decoding failed","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/datasource.go","lineNumber":289,"sourceCode":"\t\t\tif node, ok := n.OnTimerTransforms[ptransformID]; ok {\n\t\t\t\tif err := node.ProcessTimers(timerFamilyID, bcr); err != nil {\n\t\t\t\t\tlog.Warnf(ctx, \"expected transform %v to have an OnTimer method attached to handle\"+\n\t\t\t\t\t\t\"Timer Family ID: %v callback, but it did not. Please file an issue with Apache Beam\"+\n\t\t\t\t\t\t\"if you have defined OnTimer method with reproducible code at https://github.com/apache/beam/issues\", ptransformID, timerFamilyID)\n\t\t\t\t\treturn errors.WithContext(err, \"ontimer callback invocation failed\")\n\t\t\t\t}\n\t\t\t}\n\t\t\treturn nil\n\t\t})\n\n\treturn checkpoints, err\n}\n\nfunc (n *DataSource) makeReStream(ctx context.Context, cv ElementDecoder, bcr *byteCountReader, onlyStream bool) (ReStream, error) {\n\t// TODO(lostluck) 2020/02/22: Do we include the chunk size, or just the element sizes?\n\tsize, err := coder.DecodeInt32(bcr.reader)\n\tif err != nil {\n\t\treturn nil, errors.Wrap(err, \"stream size decoding failed\")\n\t}\n\n\tif onlyStream {\n\t\t// If we know the stream won't be re-iterated,\n\t\t// decode elements on demand instead to reduce memory usage.\n\t\tswitch {\n\t\tcase size >= 0:\n\t\t\treturn &singleUseReStream{\n\t\t\t\tr:    bcr,\n\t\t\t\td:    cv,\n\t\t\t\tsize: int(size),\n\t\t\t}, nil\n\t\tcase size == -1:\n\t\t\treturn &singleUseMultiChunkReStream{\n\t\t\t\tr: bcr,\n\t\t\t\td: cv,\n\t\t\t\topen: func(bcr *byteCountReader) (Stream, error) {\n\t\t\t\t\ttokenLen, err := coder.DecodeVarInt(bcr.reader)","sourceCodeStart":271,"sourceCodeEnd":307,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/datasource.go#L271-L307","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Check the upstream writer's coder matches the reader's expectation","Look for earlier decode errors in the log that would explain stream misalignment","Re-run the failing bundle; transient runner data issues may resolve","Verify no custom coder writes a different stream framing than the standard size-prefixed format"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// stream framing errors are not retryable in user code; rely on runner retry\nif strings.Contains(err.Error(), \"stream size decoding failed\") {\n    log.Errorf(\"corrupt stream framing: %v\", err)\n    return err\n}","preventionTips":["Verify coder consistency between writer and reader stages","Monitor shuffle/runner logs for truncation","Avoid custom stream framing that deviates from Beam's size-prefixed format"],"tags":["beam","go","decoding","stream"],"backgroundTag":"unexpected-response-shape","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}