{"record":{"id":"195d7bd0b28686e6","repo":"risingwavelabs/risingwave","slug":"left-barrier-received-while-right-stream-end","errorCode":null,"errorMessage":"left barrier received while right stream end","messagePattern":"left barrier received while right stream end","errorType":"error_code","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/barrier_align.rs","lineNumber":102,"sourceCode":"                        }\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                }\n                break;\n            }\n            Either::Left((Some(msg), _)) => match msg? {\n                Message::Watermark(watermark) => yield AlignedMessage::WatermarkLeft(watermark),\n                Message::Chunk(chunk) => yield AlignedMessage::Left(chunk),\n                Message::Barrier(barrier) => loop {\n                    let start_time = Instant::now();\n                    // received left barrier, waiting for right barrier\n                    match right\n                        .next()\n                        .instrument_await(await_tree::span!(\n                            \"barrier_align_wait_right epoch={}\",\n                            barrier.epoch.curr\n                        ))\n                        .await","sourceCodeStart":84,"sourceCodeEnd":120,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/barrier_align.rs#L84-L120","documentation":"Mirror case of the right-side error: after the left input ended, the right stream delivered a Barrier. The barrier aligner cannot forward a barrier to a downstream that expects both sides aligned, so it errors out.","triggerScenarios":"The right input keeps producing barriers after the left input stream returned None, in `barrier_align` (exercised via `barrier_align_for_test`).","commonSituations":"Asymmetric upstream failure/shutdown on one side of a binary operator; test streams with mismatched message sequences; one upstream actor terminated early.","solutions":["Make both barrier-producing upstreams terminate in lockstep (same barriers, same shutdown).","Investigate why the left upstream ended early — usually an upstream failure visible in earlier logs.","In tests, ensure both input streams contain barriers at the same positions."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// ensure both test inputs end together\nassert!(right_msgs.iter().rposition(|m| m.is_barrier()).unwrap_or(0) <= left_msgs.len());","typeGuard":null,"tryCatchPattern":"match barrier_align(left, right).next().await { Some(Err(e)) if e.to_string().contains(\"left barrier received while right stream end\") => { align_inputs(); } other => ... }","preventionTips":["Terminate both barrier streams in lockstep","Propagate single-side upstream failure to the other side","Construct test inputs with barrier sequences aligned 1:1"],"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"}