{"record":{"id":"22f76310e3477774","repo":"apache/beam","slug":"received-stream-with-marker-size-of-d","errorCode":null,"errorMessage":"received stream with marker size of %d","messagePattern":"received stream with marker size of (.+?)","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/datasource.go","lineNumber":385,"sourceCode":"\t\t\t\t\tnext: &proxyReStream{\n\t\t\t\t\t\topen: func() (Stream, error) {\n\t\t\t\t\t\t\tr, err := n.state.OpenIterable(ctx, n.SID, token)\n\t\t\t\t\t\t\tif err != nil {\n\t\t\t\t\t\t\t\treturn nil, err\n\t\t\t\t\t\t\t}\n\t\t\t\t\t\t\t// We can't re-use the original bcr, since we may get new iterables,\n\t\t\t\t\t\t\t// or multiple of them at the same time, but we can re-use the count itself.\n\t\t\t\t\t\t\tr = &byteCountReader{reader: r, count: bcr.count}\n\t\t\t\t\t\t\treturn &elementStream{r: r, ec: cv}, nil\n\t\t\t\t\t\t},\n\t\t\t\t\t},\n\t\t\t\t}, nil\n\t\t\tdefault:\n\t\t\t\treturn nil, errors.Errorf(\"multi-chunk stream with invalid chunk size of %d\", chunk)\n\t\t\t}\n\t\t}\n\tdefault:\n\t\treturn nil, errors.Errorf(\"received stream with marker size of %d\", size)\n\t}\n}\n\nfunc readStreamToBuffer(cv ElementDecoder, r io.Reader, size int64, buf []FullValue) ([]FullValue, error) {\n\tfor i := int64(0); i < size; i++ {\n\t\tvalue, err := cv.Decode(r)\n\t\tif err != nil {\n\t\t\treturn nil, errors.Wrap(err, \"stream value decode failed\")\n\t\t}\n\t\tbuf = append(buf, *value)\n\t}\n\treturn buf, nil\n}\n\n// FinishBundle resets the source.\nfunc (n *DataSource) FinishBundle(ctx context.Context) error {\n\tn.mu.Lock()\n\tdefer n.mu.Unlock()","sourceCodeStart":367,"sourceCodeEnd":403,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/datasource.go#L367-L403","documentation":"The first int32 in a nested stream must be a known marker: -1 (multi-chunked), 0, or a positive element count. Receiving any other value produces this errors.Errorf, meaning the stream framing is unrecognized — typically a coder/protocol mismatch.","triggerScenarios":"makeReStream's size switch falls to `default` because size is neither -1, 0, nor >0 (e.g. -2 or another negative value).","commonSituations":"Runner and SDK disagree on stream encoding (version skew); custom element coders emitting stray bytes that shift the size marker; data corruption in transit.","solutions":["Align SDK versions between pipeline submission and workers","Audit any custom coders for extra/missing bytes that offset the stream","Log the offending size value and compare against the expected framing (-1, 0, or N)","Check runner-side data serialization for recent changes"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// validate stream marker before handing off to Beam internals\n// marker must be -1, 0, or a positive count\nif !(size == -1 || size >= 0) { /* impossible for int32, but for custom formats: */ }\n// for custom encoders: assert recognized marker\nif size != -1 && size < 0 {\n    return fmt.Errorf(\"unsupported stream marker %d\", size)\n}","typeGuard":null,"tryCatchPattern":"if strings.Contains(err.Error(), \"marker size of\") {\n    return fmt.Errorf(\"unrecognized stream framing: %w\", err)\n}","preventionTips":["Pin Beam SDK versions across pipeline and workers","Audit custom coders for byte misalignment","Verify the writer uses the standard size-marker stream format"],"tags":["beam","go","wire-format","version-skew"],"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-14T21:17:11.552Z"}