{"record":{"id":"27c2d393c244958a","repo":"apache/beam","slug":"broken-stream-v-iter","errorCode":null,"errorMessage":"broken stream: %v","messagePattern":"broken stream: (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"sdks/go/pkg/beam/register/iter.go","lineNumber":63,"sourceCode":"func (v *iter1[T]) Value() any {\n\treturn v.invoke\n}\n\nfunc (v *iter1[T]) 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\nfunc (v *iter1[T]) invoke(value *T) 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\t*value = elm.Elm.(T)\n\treturn true\n}\n\ntype iter2[T1, T2 any] struct {\n\ts exec.ReStream\n\n\t// cur is the \"current\" stream, if any.\n\tcur exec.Stream\n}\n\nfunc (v *iter2[T1, T2]) Init() error {\n\tcur, err := v.s.Open()\n\tif err != nil {\n\t\treturn err\n\t}\n\tv.cur = cur","sourceCodeStart":45,"sourceCodeEnd":81,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/register/iter.go#L45-L81","documentation":"iter1[T].invoke reads the next element from the side-input/reiteration stream; EOF cleanly ends iteration (returns false), but any other read error triggers panic(\"broken stream: %v\", err). This signals the element stream (typically a gRPC-backed reader between stages) failed mid-iteration, so iteration cannot continue safely.","triggerScenarios":"Iterating a side input or grouped input where cur.Read() returns a non-EOF error: the remote data channel broke, deserialization failed, or the upstream writer died mid-stream.","commonSituations":"Worker-to-worker gRPC side-input reads over unstable networks; runner-side bugs truncating streams; large side inputs hitting connection/data limits in Dataflow or Flink runners.","solutions":["Check the wrapped error for the transport/decode failure and retry the pipeline/workers.","Reduce side-input size (filter earlier) or switch to a CoGBK/GBK join to avoid huge side inputs.","Verify runner/SDK version compatibility (mismatched beam versions can break the wire protocol).","If transient network issues, increase runner-level retry/network timeouts."],"exampleFix":"// before\nfor v.Range(&x) {\n    process(x) // panics mid-iteration on broken stream\n}\n\n// after\n// runner-level: pin matching SDK/runner versions and shrink side inputs\nside := beam.SideInput{Input: beam.CombinePerKey(sum, bigPColl)}\nsmall := beam.ParDo(s, filteredFn, smallPColl) // filter before side input","handlingStrategy":"retry","validationCode":"// no pre-call validation possible; ensure side inputs are small and versions match","typeGuard":null,"tryCatchPattern":"defer func() {\n    if r := recover(); r != nil {\n        if s, ok := r.(string); ok && strings.HasPrefix(s, \"broken stream\") {\n            log.Printf(\"side input stream broken: %s\", s) // trigger runner retry\n        }\n    }\n}()","preventionTips":["Keep SDK and runner versions in lockstep to avoid wire-protocol breaks.","Filter/reduce side inputs before use to shrink streamed data.","Enable runner-level retries for transient worker network failures."],"tags":["go","apache-beam","stream","side-input","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"}