{"record":{"id":"644387c43298310d","repo":"apache/beam","slug":"source-decode-failed","errorCode":null,"errorMessage":"source decode failed","messagePattern":"source decode failed","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/datasource.go","lineNumber":229,"sourceCode":"\n\thasSplit := map[string]bool{}\n\tvar checkpoints []*Checkpoint\n\terr := n.process(ctx, func(bcr *byteCountReader, ptransformID string) error {\n\t\t// Check if this transform has already successfully, and if so, skip reading and decoding of the elements in the buffer.\n\t\tif hasSplit[ptransformID] {\n\t\t\treturn nil\n\t\t}\n\t\tfor {\n\t\t\t// TODO(lostluck) 2020/02/22: Should we include window headers or just count the element sizes?\n\t\t\tws, t, pn, err := DecodeWindowedValueHeader(wc, bcr.reader)\n\t\t\tif err != nil {\n\t\t\t\treturn err\n\t\t\t}\n\n\t\t\t// Decode key or parallel element.\n\t\t\tpe, err := cp.Decode(bcr)\n\t\t\tif err != nil {\n\t\t\t\treturn errors.Wrap(err, \"source decode failed\")\n\t\t\t}\n\t\t\tpe.Timestamp = t\n\t\t\tpe.Windows = ws\n\t\t\tpe.Pane = pn\n\n\t\t\tvar valReStreams []ReStream\n\t\t\tfor _, cv := range cvs {\n\t\t\t\tvalues, err := n.makeReStream(ctx, cv, bcr, len(cvs) == 1 && n.singleIterate)\n\t\t\t\tif err != nil {\n\t\t\t\t\treturn err\n\t\t\t\t}\n\t\t\t\tvalReStreams = append(valReStreams, values)\n\t\t\t}\n\n\t\t\tif err := n.Out.ProcessElement(ctx, pe, valReStreams...); err != nil {\n\t\t\t\treturn err\n\t\t\t}\n\t\t\t// Collect the actual size of the element, and reset the bytecounter reader.","sourceCodeStart":211,"sourceCodeEnd":247,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/datasource.go#L211-L247","documentation":"Raised when decoding a key or parallel element from the incoming data stream fails: cp.Decode(bcr) returns an error while DataSource.invokeProcess unfolds the element. It is wrapped as 'source decode failed', meaning the coder could not interpret the byte stream for this element.","triggerScenarios":"A coder (cp, an ElementDecoder from the plan) tries to decode bytes read from bcr and returns an error — wrong coder, truncated payload, or malformed bytes.","commonSituations":"Mismatched coder versions after a pipeline update; custom coders whose Encode/Decode are not symmetric; corrupted or truncated element buffers from the runner; cross-version serialized data.","solutions":["Verify the coder registered for the PTransform matches the data actually sent","Check for pipeline version drift between writer and reader stages; use consistent SDK versions","Add debugging output to the custom coder's Decode to see which bytes fail","Regenerate/recompile custom coders if their schema changed"],"exampleFix":"// before\nfunc (c *myCoder) Decode(r io.Reader) (interface{}, error) {\n    var b [8]byte\n    io.ReadFull(r, b[:]) // length not validated\n    return string(b[:]), nil\n}\n// after\nfunc (c *myCoder) Decode(r io.Reader) (interface{}, error) {\n    l, err := coder.DecodeVarInt(r)\n    if err != nil { return nil, err }\n    buf := make([]byte, l)\n    if _, err := io.ReadFull(r, buf); err != nil { return nil, err }\n    return string(buf), nil\n}","handlingStrategy":"validation","validationCode":"// ensure custom coder round-trips before deploying\ndef roundTrip(c ElementCoder, v interface{}) bool {\n    var buf bytes.Buffer\n    if err := c.Encode(v, &buf); err != nil { return false }\n    got, err := c.Decode(bytes.NewReader(buf.Bytes()))\n    return err == nil && reflect.DeepEqual(got, v)\n}","typeGuard":"func isDecodeError(err error) bool {\n    return strings.Contains(fmt.Sprint(err), \"source decode failed\")\n}","tryCatchPattern":"if err != nil {\n    var decErr = \"source decode failed\"\n    if strings.Contains(err.Error(), decErr) {\n        // route to dead-letter / quarantine the record\n    }\n}","preventionTips":["Always unit-test custom coder round-trips","Keep SDK versions identical between pipeline submit and workers","Avoid changing custom coder formats without a migration plan","Log failing byte offsets in custom decoders"],"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"}