{"record":{"id":"ea569a9c49456bf7","repo":"apache/beam","slug":"broken-stream","errorCode":null,"errorMessage":"broken stream","messagePattern":"broken stream","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"sdks/go/pkg/beam/core/runtime/exec/input.go","lineNumber":163,"sourceCode":"\t\tpanic(\"Init() not called\")\n\t}\n\tif err := v.cur.Close(); err != nil {\n\t\treturn err\n\t}\n\tv.cur = nil\n\treturn nil\n}\n\nfunc (v *iterValue) invoke(args []reflect.Value) []reflect.Value {\n\tif v.cur == nil {\n\t\tpanic(\"Init() not called\")\n\t}\n\telm, err := v.cur.Read()\n\tif err != nil {\n\t\tif err == io.EOF {\n\t\t\treturn []reflect.Value{reflect.ValueOf(false)}\n\t\t}\n\t\tpanic(errors.Wrap(err, \"broken stream\"))\n\t}\n\n\t// We expect 1-3 out parameters: func (*int, *string) bool.\n\n\tisKey := true\n\tfor i, t := range v.types {\n\t\tvar v reflect.Value\n\t\tswitch {\n\t\tcase isKey:\n\t\t\tv = reflect.ValueOf(Convert(elm.Elm, t))\n\t\t\tisKey = false\n\t\tdefault:\n\t\t\tv = reflect.ValueOf(Convert(elm.Elm2, t))\n\t\t}\n\t\targs[i].Elem().Set(v)\n\t}\n\treturn []reflect.Value{reflect.ValueOf(true)}\n}","sourceCodeStart":145,"sourceCodeEnd":181,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/input.go#L145-L181","documentation":"invoke reads the next element from the reusable re-iterator backing a DoFn's main input; io.EOF means the iteration is done (returns false), but any other read failure is fatal and is paniced as 'broken stream'. The runtime deliberately panics because a broken reader means the bundle cannot continue producing correct results.","triggerScenarios":"invoke() on a reusableReader where v.cur.Read() returns a non-EOF error: the side data plane stream delivering elements failed mid-iteration (transport error, decode corruption, closed reader).","commonSituations":"Network failure between runner and worker during element delivery; an upstream decode error (e.g. 4995/4996) wrapped and re-paniced here; bundle cancellation closing the stream unexpectedly.","solutions":["Let the runner retry the bundle; look at the wrapped cause in the panic chain for the root transport or decode error.","Check runner-to-worker connectivity and data-plane health.","If caused by a decode mismatch, fix the coder/element type alignment feeding this DoFn.","Inspect whether user code closed or consumed the iterator concurrently."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// user code wrapping the iterator should not swallow non-EOF errors\nfor {\n    v, err := iter.Read()\n    if err == io.EOF { break }\n    if err != nil { return fmt.Errorf(\"broken stream: %w\", err) /* let runner retry */ }\n}","preventionTips":["Never close or share the DoFn's iterator concurrently with the framework.","Check the panic's wrapped cause for the true transport/decode error.","Ensure worker-runner connectivity is reliable in the cluster setup."],"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"}