{"record":{"id":"5d5077780d30338c","repo":"risingwavelabs/risingwave","slug":"log-store-rewind-start-epoch-not-later-than-fir","errorCode":null,"errorMessage":"log_store_rewind_start_epoch {} not later than first_epoch {}","messagePattern":"log_store_rewind_start_epoch (.+?) not later than first_epoch (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/coordinate.rs","lineNumber":87,"sourceCode":"        })\n    }\n}\n\n#[async_trait]\nimpl<W: SinkWriter<CommitMetadata = Option<SinkMetadata>>> LogSinker for CoordinatedLogSinker<W> {\n    async fn consume_log_and_sink(self, mut log_reader: impl SinkLogReader) -> Result<!> {\n        let (mut coordinator_stream_handle, log_store_rewind_start_epoch) = self\n            .sink_coordinate_client\n            .new_stream_handle(&self.param, self.vnode_bitmap)\n            .await?;\n        let mut sink_writer = self.writer;\n        log_reader.start_from(log_store_rewind_start_epoch).await?;\n        let mut first_item = log_reader.next_item().await?;\n        if let (Some(log_store_rewind_start_epoch), (first_epoch, _)) =\n            (log_store_rewind_start_epoch, &first_item)\n        {\n            if log_store_rewind_start_epoch >= *first_epoch {\n                bail!(\n                    \"log_store_rewind_start_epoch {} not later than first_epoch {}\",\n                    log_store_rewind_start_epoch,\n                    first_epoch\n                );\n            }\n        } else {\n            let &(initial_epoch, _) = &first_item;\n            let aligned_initial_epoch = coordinator_stream_handle\n                .align_initial_epoch(initial_epoch)\n                .await?;\n            if initial_epoch != aligned_initial_epoch {\n                warn!(\n                    initial_epoch,\n                    aligned_initial_epoch,\n                    sink_id = %self.param.sink_id,\n                    \"initial epoch not matched aligned initial epoch\"\n                );\n                let mut peeked_first = Some(first_item);","sourceCodeStart":69,"sourceCodeEnd":105,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/coordinate.rs#L69-L105","documentation":"After rewinding the sink's log store reader to log_store_rewind_start_epoch, the coordinator expects the first log item's epoch to be strictly later than the rewind start epoch. If the first item's epoch is >= the rewind start, the rewind semantics are violated and the log would be consumed from a point at or before the requested epoch, so it bails.","triggerScenarios":"consume_log_and_sink is called with a rewind start epoch, and the log reader's first returned item carries first_epoch such that log_store_rewind_start_epoch >= first_epoch (rewind point not strictly before the first available item).","commonSituations":"Recovery after failover where the requested rewind epoch equals the first epoch still present in the log store; stale log store state after migration or incorrect epoch bookkeeping in the meta log store.","solutions":["Check the log store to confirm the rewind start epoch is strictly less than the first retained epoch.","Retrigger recovery with a corrected rewind start epoch (usually the last checkpoint epoch).","Inspect meta log store cleanup/retention to ensure it did not trim past the requested epoch."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"if let Some(start) = log_store_rewind_start_epoch {\n    if start >= first_epoch_in_log {\n        // adjust rewind point to last checkpoint epoch before retrying\n    }\n}","typeGuard":null,"tryCatchPattern":"match result {\n    Err(e) if e.to_string().contains(\"log_store_rewind_start_epoch\") => retry_with_last_checkpoint_epoch(),\n    other => other,\n}","preventionTips":["Always rewind from the last completed checkpoint epoch, never an arbitrary epoch.","Monitor log store retention so epochs requested for rewind are not trimmed."],"tags":["rust","risingwave","log-store","recovery"],"backgroundTag":"invalid-argument-value","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"}