{"record":{"id":"52c962b5b3a1100e","repo":"risingwavelabs/risingwave","slug":"current-epoch-has-exceeded-the-epoch-of-the","errorCode":null,"errorMessage":"current epoch {} has exceeded the epoch {} of the stream that has not started","messagePattern":"current epoch (.+?) has exceeded the epoch (.+?) of the stream that has not started","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/kv_log_store/serde.rs","lineNumber":972,"sourceCode":"            }\n        };\n\n        if prev_epoch >= epoch {\n            return Err(anyhow!(\n                \"epoch {} should be greater than prev epoch {}\",\n                epoch,\n                prev_epoch\n            ));\n        }\n\n        while let Some((stream_epoch, _)) = self.not_started_streams.last() {\n            if *stream_epoch > epoch {\n                // Current epoch has not reached the first epoch of\n                // the stream. Later streams must also have greater epoch, so break here.\n                break;\n            }\n            if *stream_epoch < epoch {\n                return Err(anyhow!(\n                    \"current epoch {} has exceeded the epoch {} of the stream that has not started\",\n                    epoch,\n                    stream_epoch\n                ));\n            }\n            let (_, stream) = self.not_started_streams.pop().expect(\"should not be empty\");\n            self.row_streams.push(stream.into_future());\n        }\n\n        self.stream_state = StreamState::AllConsumingRow { curr_epoch: epoch };\n        Ok(())\n    }\n\n    async fn next_op(&mut self) -> LogStoreResult<Option<AlignedLogStoreRow>> {\n        while let (Some(result), stream) = self\n            .row_streams\n            .next()\n            .await","sourceCodeStart":954,"sourceCodeEnd":990,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/kv_log_store/serde.rs#L954-L990","documentation":"Thrown while draining `not_started_streams` during barrier validation: the incoming epoch has skipped past the epoch recorded for a stream that had not yet started, so its bookkeeping entry can never be matched. The log store tracks streams whose first epoch is in the future; if a barrier arrives with an epoch greater than such a stream's first epoch, the stream's state is inconsistent and cannot be reconciled.","triggerScenarios":"In `validate_epoch`, when popping `not_started_streams` and finding `*stream_epoch < epoch` — a barrier with epoch E arrives for a stream whose recorded first epoch is smaller than E (epoch skipped over it).","commonSituations":"A stream was registered with an expected start epoch but its data/barriers were processed under a later epoch (actor creation delayed, backfill altered epochs), or meta node updated the stream epoch list inconsistently with barrier flow.","solutions":["Verify that a stream's registered first epoch matches the epoch of the first barrier it actually receives.","Check barrier flow so streams that have not started are not skipped by later epochs; resume/start them before the epoch advances past their registration.","Confirm meta node stream-building completion ordering: a stream must start before barriers with greater epochs are validated.","Log `not_started_streams` contents when the error fires to identify which stream's epoch was skipped."],"exampleFix":"// before: barrier epoch advanced past not-started stream\nstream_epoch = 100; barrier_epoch = 200; // 100 < 200 -> error\n// after: ensure stream starts at/before barrier epoch\nstream_epoch = 200; barrier_epoch = 200; // equal -> popped and matched\n","handlingStrategy":"validation","validationCode":"// Rust: before validating a barrier, ensure no not-started stream would be skipped\nfn would_skip_epoch(not_started_streams: &[u64], epoch: u64) -> bool {\n    not_started_streams.iter().any(|s| s < &epoch)\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Register streams with the exact epoch of their first barrier.","Ensure stream building/creation completes before barriers of later epochs are processed.","Monitor creation-to-first-barrier latency to catch streams stuck in not-started state."],"tags":["streaming","epoch","barrier","stream-lifecycle"],"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"}