{"record":{"id":"0ac3a4a5dde892e1","repo":"risingwavelabs/risingwave","slug":"right-barrier-received-while-left-stream-end","errorCode":null,"errorMessage":"right barrier received while left stream end","messagePattern":"right barrier received while left stream end","errorType":"error_code","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/barrier_align.rs","lineNumber":87,"sourceCode":"        let select_result = if prefer_left {\n            select(left.next(), right.next()).await\n        } else {\n            match select(right.next(), left.next()).await {\n                Either::Left(x) => Either::Right(x),\n                Either::Right(x) => Either::Left(x),\n            }\n        };\n        match select_result {\n            Either::Left((None, _)) => {\n                // left stream end, passthrough right chunks\n                while let Some(msg) = right.next().await {\n                    match msg? {\n                        Message::Watermark(watermark) => {\n                            yield AlignedMessage::WatermarkRight(watermark)\n                        }\n                        Message::Chunk(chunk) => yield AlignedMessage::Right(chunk),\n                        Message::Barrier(_) => {\n                            bail!(\"right barrier received while left stream end\");\n                        }\n                    }\n                }\n                break;\n            }\n            Either::Right((None, _)) => {\n                // right stream end, passthrough left chunks\n                while let Some(msg) = left.next().await {\n                    match msg? {\n                        Message::Watermark(watermark) => {\n                            yield AlignedMessage::WatermarkLeft(watermark)\n                        }\n                        Message::Chunk(chunk) => yield AlignedMessage::Left(chunk),\n                        Message::Barrier(_) => {\n                            bail!(\"left barrier received while right stream end\");\n                        }\n                    }\n                }","sourceCodeStart":69,"sourceCodeEnd":105,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/barrier_align.rs#L69-L105","documentation":"In `barrier_align`, after the right input stream ended, the left stream delivered a Barrier. Barrier alignment requires both inputs to see barriers in lockstep; a barrier on a stream whose counterpart already ended breaks the alignment protocol, so it aborts.","triggerScenarios":"Two-input alignment where one input finishes while the other still emits barriers — e.g. mismatched upstream lifetimes feeding a binary executor, or a test harness (`barrier_align_for_test`) with asymmetric stream ends.","commonSituations":"Binary operators (join/union) whose upstreams terminate at different times; network failure ending one upstream early; incorrectly constructed test streams.","solutions":["Ensure both upstreams feeding the aligned executor have the same lifetime and are torn down together.","Check for upstream failures that closed one side prematurely (look for earlier errors on that actor).","Fix test/executor wiring so barrier streams are not terminated independently."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// ensure upstreams share lifetime: end both streams at the same barrier\nassert_eq!(left.barrier_count(), right.barrier_count());","typeGuard":null,"tryCatchPattern":"match barrier_align(left, right).next().await { Some(Err(e)) if e.to_string().contains(\"received while\") => { tear_down_both(); } other => ... }","preventionTips":["Wire binary-operator upstreams so they are dropped together","Fail both sides when either upstream fails","Keep test streams symmetric with matching barriers"],"tags":["streaming","barrier","alignment","invariant"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}