{"record":{"id":"605a3ae1e5c0a495","repo":"apache/beam","slug":"broken-stream-v-inputs","errorCode":null,"errorMessage":"broken stream: %v","messagePattern":"broken stream: (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/core/runtime/exec/optimized/inputs.tmpl","lineNumber":75,"sourceCode":"\treturn v.fn\n}\n\nfunc (v *iterNative) Reset() error {\n\tif err := v.cur.Close(); err != nil {\n\t\treturn err\n\t}\n\tv.cur = nil\n\treturn nil\n}\n\n{{range $x := .X}}\nfunc (v *iterNative) read{{$x.Name}}(val *{{$x.Type}}) bool {\n\telm, err := v.cur.Read()\n\tif err != nil {\n\t\tif err == io.EOF {\n\t\t\treturn false\n\t\t}\n\t\tpanic(fmt.Sprintf(\"broken stream: %v\", err))\n\t}\n    *val = elm.Elm.({{$x.Type}})\n    return true\n}\n\nfunc iterMaker{{$x.Name}}(s exec.ReStream) exec.ReusableInput {\n\tret := &iterNative{s: s}\n\tret.fn = ret.read{{$x.Name}}\n\treturn ret\n}\n\n{{range $y := .Y}}\nfunc (v *iterNative) read{{$x.Name}}{{$y.Name}}(key *{{$x.Type}}, value *{{$y.Type}}) bool {\n\telm, err := v.cur.Read()\n\tif err != nil {\n\t\tif err == io.EOF {\n\t\t\treturn false\n\t\t}","sourceCodeStart":57,"sourceCodeEnd":93,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/optimized/inputs.tmpl#L57-L93","documentation":"This generated single-value iterator reads elements from a runtime stream (FullValue stream). io.EOF is the normal end-of-stream signal; any other read error means the data channel broke mid-stream, so the generated code panics because iterating cannot continue meaningfully. It surfaces as a crash during element iteration in an optimized pipeline step.","triggerScenarios":"A side-input or GBK-style ReStream's underlying reader returns a non-EOF error (broken gRPC data channel, source failure, deserialization error) while readX is pulling the next element.","commonSituations":"Bundle failures mid-stream — e.g., the runner kills the data connection, network interruption to a remote runner, or an upstream source error corrupting the stream.","solutions":["Check bundle/worker logs for the root cause of the stream failure (network errors, OOM, source failure)","Retry the pipeline/job — broken streams are often transient infrastructure failures","If reproducible, inspect the upstream source/transform producing the stream for errors, and upgrade Beam to get improved stream error propagation"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"defer func() {\n    if r := recover(); r != nil {\n        if strings.HasPrefix(fmt.Sprint(r), \"broken stream:\") {\n            log.Printf(\"stream broke during iteration: %v — retry bundle\", r)\n        }\n        panic(r)\n    }\n}()","preventionTips":["Retry failed bundles/pipelines; broken streams are often transient","Monitor network stability between worker and runner","Upgrade Beam for better stream error propagation","Check upstream sources for failures before debugging the consumer"],"tags":["go","apache-beam","stream","panic"],"backgroundTag":"broken-pipe","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"}