{"record":{"id":"391df4e270695af7","repo":"risingwavelabs/risingwave","slug":"when-any-of-the-stream-reaches-the-end-other-stre","errorCode":null,"errorMessage":"when any of the stream reaches the end, other stream should also reaches the end, but poll result: {:?}","messagePattern":"when any of the stream reaches the end, other stream should also reaches the end, but poll result: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/stream/src/common/log_store_impl/kv_log_store/serde.rs","lineNumber":1133,"sourceCode":"                return Err(anyhow!(\n                    \"when any of the stream reaches the end, it should be right after emitting an barrier. Current state: {:?}\",\n                    s\n                ));\n            }\n        }\n        assert!(\n            self.barrier_streams.is_empty(),\n            \"should not have any pending barrier received stream after barrier emit\"\n        );\n        if !self.not_started_streams.is_empty() {\n            return Err(anyhow!(\n                \"a stream has reached the end but some other stream has not started yet\"\n            ));\n        }\n        if cfg!(debug_assertions) {\n            while let Some((opt, _stream)) = self.row_streams.next().await {\n                if let Some(result) = opt {\n                    return Err(anyhow!(\n                        \"when any of the stream reaches the end, other stream should also reaches the end, but poll result: {:?}\",\n                        result\n                    ));\n                }\n            }\n        }\n        Ok(None)\n    }\n}\n\n#[cfg(test)]\nmod tests {\n    use std::future::poll_fn;\n    use std::iter::once;\n    use std::sync::Arc;\n    use std::task::Poll;\n\n    use bytes::Bytes;","sourceCodeStart":1115,"sourceCodeEnd":1151,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/kv_log_store/serde.rs#L1115-L1151","documentation":"In debug builds only (`cfg!(debug_assertions)`), after a stream reaches the end the KV log store reader drains the remaining row streams and expects every one to return `None` (already ended). If any stream still yields a row (`Some(result)`), streams ended at different points — contradicting the invariant that all streams end together right after a shared barrier. This is a debug-time consistency check for multi-stream replay correctness.","triggerScenarios":"In debug builds, draining the reader after end-of-stream while at least one row stream still polls to `Some(Ok(row))` — i.e. one stream kept producing rows after another had terminated.","commonSituations":"Mixed-version clusters where one writer continues past a barrier another stopped at; bugs in barrier alignment across parallel streams; manually rewritten or patched Hummock data in test/dev clusters.","solutions":["Reproduce in a debug build, then dump the per-stream rows around the final barrier epoch and check barrier alignment of the upstream streams.","Recover all streams from a common snapshot/barrier so they restart from the same epoch and end together.","Align writer versions across nodes (rolling upgrade with mismatched serde/log-store versions can misalign barriers).","If seen only in release builds (check disabled), still treat as a correctness bug and file it with repro data."],"exampleFix":"// before: streams independent, can end at different epochs\nStreamA::write(rows).await;\nStreamB::write(rows).await; // continued past StreamA's end\n// after: route all writes through the shared barrier-aligned writer\nlog_store.write_all_aligned(vec![rows_a, rows_b]).await; // ends together after barrier","handlingStrategy":"try-catch","validationCode":"// After end-of-stream, drain in a controlled way and log any laggard stream instead of failing hard\nwhile let Some((opt, stream)) = row_streams.next().await {\n    if opt.is_some() { tracing::warn!(?stream, \"stream outlived end-of-stream\"); }\n}","typeGuard":"// Rust\nfn all_streams_ended(results: &[Option<PollResult>]) -> bool {\n    results.iter().all(|r| r.is_none())\n}","tryCatchPattern":"// Rust\nmatch replay_to_end().await {\n    Err(e) if e.to_string().contains(\"other stream should also reaches the end\") => {\n        recover_from_common_barrier().await?; // re-align streams, then replay again\n    }\n    other => other,\n}","preventionTips":["Align barrier emission so all parallel streams write barriers at the same epoch","Avoid rolling upgrades that mix writer versions with different barrier cadences","Never hand-patch Hummock data in dev/test clusters","Run replay consistency tests in debug builds before release"],"tags":["streaming","log-store","debug-assert","barrier-alignment"],"backgroundTag":"internal-invariant-violation","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}