{"record":{"id":"2a89221746f6956e","repo":"apache/beam","slug":"decodestream-value-decode-failed-on-close","errorCode":null,"errorMessage":"decodeStream value decode failed on close","messagePattern":"decodeStream value decode failed on close","errorType":"error_code","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/fullvalue.go","lineNumber":237,"sourceCode":"\tr          io.Reader\n\td          ElementDecoder\n\tnext, size int\n\tret        FullValue\n}\n\n// Close causes subsequent calls to Read to return io.EOF, and drains the remaining element count\n// from the reader.\nfunc (s *decodeStream) Close() error {\n\t// On close, if next != size, we must iterate through the rest of the decoding\n\t// until the reader is drained. Otherwise we corrupt the read for the next element.\n\t//\n\t// TODO(https://github.com/apache/beam/issues/22901):\n\t// Optimize the case where we have length prefixed values\n\t// so we can avoid allocating the values in the first place.\n\tfor s.next < s.size {\n\t\terr := s.d.DecodeTo(s.r, &s.ret)\n\t\tif err != nil {\n\t\t\treturn errors.Wrap(err, \"decodeStream value decode failed on close\")\n\t\t}\n\t\ts.next++\n\t}\n\ts.r = nil\n\ts.d = nil\n\ts.ret = FullValue{}\n\treturn nil\n}\n\n// Read produces the next value in the stream.\nfunc (s *decodeStream) Read() (*FullValue, error) {\n\tif s.r == nil || s.next == s.size {\n\t\treturn nil, io.EOF\n\t}\n\terr := s.d.DecodeTo(s.r, &s.ret)\n\tif err != nil {\n\t\tif err == io.EOF {\n\t\t\treturn nil, io.EOF","sourceCodeStart":219,"sourceCodeEnd":255,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/fullvalue.go#L219-L255","documentation":"During Close of a length-prefixed decode stream, any remaining undecoded values within the declared size are drained and decoded; a decode failure there is wrapped as 'decodeStream value decode failed on close'. This means the serialized bytes on the stream are not decodable with the expected coder, surfacing at stream shutdown rather than during normal reads.","triggerScenarios":"Close() on the decode stream while s.next < s.size and s.d.DecodeTo(s.r, &s.ret) fails: truncated/corrupted runner-side data stream, coder mismatch between writer and reader, or a partially delivered gRPC element stream that ends mid-record.","commonSituations":"Runner/data-plane instability dropping bytes mid-stream; mismatched coder versions between SDK workers (e.g. pipeline built with a coder the executing worker doesn't support); custom coder emitting malformed bytes.","solutions":["Retry the bundle/stage; transient data-plane corruption usually resolves on rerun.","Verify all workers run the same SDK version as the pipeline was expanded with (coder compatibility).","Inspect custom coders/encoders registered for the element type for malformed output.","Check runner logs for transport errors (broken pipe, cancelled gRPC stream) preceding this failure."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// runner-side: check the wrapped cause\nif strings.Contains(err.Error(), \"decodeStream value decode failed on close\") {\n    log.Warnf(\"data stream corruption at close; retrying bundle: %v\", errors.Unwrap(err))\n    return retryable // surface as retryable to the runner\n}","preventionTips":["Pin all workers to the same SDK version as pipeline expansion.","Round-trip test any custom coders.","Monitor data-plane/gRPC transport errors in runner logs."],"tags":["go","apache-beam","decoding","stream"],"backgroundTag":"stream-decode-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"}