{"record":{"id":"5bf59f029d3d5d9d","repo":"apache/beam","slug":"stream-value-decode-failed","errorCode":null,"errorMessage":"stream value decode failed","messagePattern":"stream value decode failed","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/datasource.go","lineNumber":393,"sourceCode":"\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()\n\tn.source = nil\n\tn.splitIdx = 0 // Ensure errors are returned for split requests if this plan is re-used.\n\treturn n.Out.FinishBundle(ctx)\n}\n\n// Down resets the source.\nfunc (n *DataSource) Down(ctx context.Context) error {\n\tn.source = nil","sourceCodeStart":375,"sourceCodeEnd":411,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/datasource.go#L375-L411","documentation":"readStreamToBuffer decodes `size` individual elements into a buffer for a fixed-size stream; any cv.Decode failure on an element is wrapped as 'stream value decode failed'. The cause is usually a per-element decode error inside the iterable (bad bytes or an unsuitable coder).","triggerScenarios":"While reading N elements of a sized stream, element i's cv.Decode(r) returns an error (EOF mid-element, invalid encoding, custom coder failure).","commonSituations":"GBK results with elements encoded by a mismatched coder; truncated stream ending before all N elements are read; custom coders failing on particular values (e.g. invalid UTF-8, out-of-range enums).","solutions":["Fix the element coder so Encode/Decode round-trips all values","Check the stream actually contains `size` complete elements (truncation)","Test the coder in isolation with representative values","Ensure the writer and reader agree on element types"],"exampleFix":"// before\ntype enumCoder struct{}\nfunc (enumCoder) Encode(v interface{}, w io.Writer) error {\n    return coder.EncodeInt32(w, int32(v.(myEnum)), coder.EndOfLengthPrefix) // unvalidated\n}\n// after\ntype enumCoder struct{}\nfunc (enumCoder) Encode(v interface{}, w io.Writer) error {\n    e, ok := v.(myEnum)\n    if !ok { return fmt.Errorf(\"expected myEnum, got %T\", v) }\n    return coder.EncodeInt32(w, int32(e), coder.EndOfLengthPrefix)\n}","handlingStrategy":"validation","validationCode":"// pre-check coder round-trip for element types used in GBK values\nfunc testCoderRoundTrip(c ElementCoder, vals []interface{}) error {\n    for _, v := range vals {\n        var buf bytes.Buffer\n        if err := c.Encode(v, &buf); err != nil { return err }\n        if _, err := c.Decode(&buf); err != nil { return err }\n    }\n    return nil\n}","typeGuard":"func isStreamValueDecodeFailure(err error) bool {\n    return strings.Contains(err.Error(), \"stream value decode failed\")\n}","tryCatchPattern":"if err != nil && strings.Contains(err.Error(), \"stream value decode failed\") {\n    // quarantine element; check for truncation (EOF) vs. encoding error\n}","preventionTips":["Round-trip test coders over representative data","Ensure the stream carries exactly `size` complete elements","Handle all enum/range values in custom coders"],"tags":["beam","go","decoding","coder"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}