{"record":{"id":"db0cad7ed5b3e4f1","repo":"risingwavelabs/risingwave","slug":"initial-epoch-greater-than-aligned-initial-epoc","errorCode":null,"errorMessage":"initial epoch {} greater than aligned initial epoch {}","messagePattern":"initial epoch (.+?) greater than aligned initial epoch (.+?)","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/coordinate.rs","lineNumber":120,"sourceCode":"                    sink_id = %self.param.sink_id,\n                    \"initial epoch not matched aligned initial epoch\"\n                );\n                let mut peeked_first = Some(first_item);\n                first_item = loop {\n                    let (epoch, item) = if let Some(peeked_first) = peeked_first.take() {\n                        peeked_first\n                    } else {\n                        log_reader.next_item().await?\n                    };\n                    match epoch.cmp(&aligned_initial_epoch) {\n                        Ordering::Less => {\n                            continue;\n                        }\n                        Ordering::Equal => {\n                            break (epoch, item);\n                        }\n                        Ordering::Greater => {\n                            return Err(anyhow!(\n                                \"initial epoch {} greater than aligned initial epoch {}\",\n                                initial_epoch,\n                                aligned_initial_epoch\n                            )\n                            .into());\n                        }\n                    }\n                };\n            }\n        }\n\n        let mut first_item = Some(first_item);\n\n        #[derive(Debug)]\n        enum LogConsumerState {\n            /// Mark that the log consumer is not initialized yet\n            Uninitialized,\n","sourceCodeStart":102,"sourceCodeEnd":138,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/coordinate.rs#L102-L138","documentation":"While consuming the log to align with the sink's initial_epoch, the coordinator compares each item's epoch with initial_epoch. If an item's epoch is Greater than the aligned initial epoch, the stream has skipped past the expected initial epoch, meaning the log and the sink state are inconsistent, so it returns an error instead of breaking at the aligned epoch.","triggerScenarios":"consume_log_and_sink with initial_epoch set; the log reader yields an item whose epoch is greater than initial_epoch before an Equal match, i.e. no log entry exactly matches the sink's initial epoch.","commonSituations":"Failover recovery where the sink's initial epoch is ahead of the log store's retained epochs; version mismatch after schema change or incorrect epoch propagation from meta during sink scaling.","solutions":["Verify the initial_epoch passed to the coordinator matches an epoch actually present in the log store.","Recover from the last valid checkpoint so initial_epoch aligns with a logged epoch.","Check meta-side epoch bookkeeping after scale-out/schema change for dropped log entries."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"if err.to_string().contains(\"greater than aligned initial epoch\") {\n    // recompute initial_epoch from the last valid checkpoint and retry recovery\n}","preventionTips":["Derive initial_epoch from meta's log store, not from sink-local state.","Avoid manual recovery epochs; always use the coordinator's checkpoint bookkeeping."],"tags":["rust","risingwave","log-store","epoch"],"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"}