{"record":{"id":"0aa7e6e9a99cf850","repo":"apache/beam","slug":"broken-stream-v","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.go","lineNumber":952,"sourceCode":"\t}\n\t*key = elm.Elm.([]byte)\n\t*value = elm.Elm2.(typex.W)\n\treturn true\n}\n\nfunc iterMakerByteSliceTypex_W(s exec.ReStream) exec.ReusableInput {\n\tret := &iterNative{s: s}\n\tret.fn = ret.readByteSliceTypex_W\n\treturn ret\n}\n\nfunc (v *iterNative) readByteSliceTypex_X(key *[]byte, value *typex.X) 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*key = elm.Elm.([]byte)\n\t*value = elm.Elm2.(typex.X)\n\treturn true\n}\n\nfunc iterMakerByteSliceTypex_X(s exec.ReStream) exec.ReusableInput {\n\tret := &iterNative{s: s}\n\tret.fn = ret.readByteSliceTypex_X\n\treturn ret\n}\n\nfunc (v *iterNative) readByteSliceTypex_Y(key *[]byte, value *typex.Y) 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":934,"sourceCodeEnd":970,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/go/pkg/beam/core/runtime/exec/optimized/inputs.go#L934-L970","documentation":"This is a generated optimized reader for byte-slice keyed iteration (key []byte, value typex.X). It reads from the underlying element stream; on any read error other than io.EOF (which signals normal end of stream) it panics with 'broken stream', wrapping the cause. A broken stream means the element channel or reader backing the iteration failed mid-stream.","triggerScenarios":"v.cur.Read() returns a non-EOF error during side-input iteration — e.g. the remote channel carrying side input data failed, decoding of an element failed, or the runner closed the stream prematurely.","commonSituations":"Network failures between runner and SDK harness while streaming side input data; runner OOM/crash truncating the side input stream; data corruption or deserialization errors on element decode.","solutions":["Read the wrapped cause in the panic message to identify whether it is transport, decode, or runner-side.","Retry the bundle/work item — transient channel failures often succeed on retry.","Check runner (e.g. Dataflow/Flink) logs for the producing stage's health and any channel closures.","If caused by large side inputs, reduce side input size or switch to a different input pattern (e.g. co-group) to avoid streaming limits.","Upgrade the SDK/runner if the error is a known deserialization regression."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"defer func() {\n\tif r := recover(); r != nil {\n\t\tif s, ok := r.(string); ok && strings.HasPrefix(s, \"broken stream\") {\n\t\t\t// surface wrapped cause and retry the work item via the runner's retry mechanism\n\t\t\tlog.Printf(\"side input stream failed: %s\", s)\n\t\t}\n\t}\n}()","preventionTips":["Keep side inputs small enough to stream reliably; prefer larger data via co-group/join.","Rely on runner-level bundle retries for transient channel failures.","Monitor network stability between runner and SDK harness workers.","Upgrade SDK/runner pairs together to avoid decode mismatches."],"tags":["go","apache-beam","stream","side-input"],"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"}