{"record":{"id":"c23ceeef5363b087","repo":"risingwavelabs/risingwave","slug":"lancedb-pre-commit-epoch-does-not-match-commit","errorCode":null,"errorMessage":"LanceDB pre-commit epoch {} does not match commit epoch {}","messagePattern":"LanceDB pre-commit epoch (.+?) does not match commit epoch (.+?)","errorType":"exception","errorClass":"SinkError::LanceDb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/lancedb.rs","lineNumber":750,"sourceCode":"            LanceDbPreCommitMetadata {\n                sink_id: self.sink_id.clone(),\n                epoch,\n                fragments,\n            }\n            .try_into_bytes()?,\n        ))\n    }\n\n    async fn commit_data(&mut self, epoch: u64, commit_metadata: Vec<u8>) -> Result<()> {\n        tracing::debug!(\"Starting LanceDB two-phase commit in epoch {epoch}.\");\n\n        if commit_metadata.is_empty() {\n            return Ok(());\n        }\n\n        let pre_commit_metadata = LanceDbPreCommitMetadata::try_from_bytes(&commit_metadata)?;\n        if pre_commit_metadata.epoch != epoch {\n            return Err(SinkError::LanceDb(anyhow!(\n                \"LanceDB pre-commit epoch {} does not match commit epoch {}\",\n                pre_commit_metadata.epoch,\n                epoch\n            )));\n        }\n        if pre_commit_metadata.sink_id != self.sink_id {\n            return Err(SinkError::LanceDb(anyhow!(\n                \"LanceDB pre-commit sink id {} does not match coordinator sink id {}\",\n                pre_commit_metadata.sink_id,\n                self.sink_id\n            )));\n        }\n\n        let transaction_properties = pre_commit_metadata.transaction_properties();\n        self.commit_fragments(\n            epoch,\n            pre_commit_metadata.fragments,\n            Some(transaction_properties),","sourceCodeStart":732,"sourceCodeEnd":768,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/lancedb.rs#L732-L768","documentation":"During commit_data, the LanceDB sink deserializes the opaque pre-commit state written in pre_commit and compares its recorded epoch to the epoch the coordinator asks to commit. A mismatch means the commit belongs to a different (typically older) epoch than the one prepared — usually the consequence of a recovery/restart where a stale prepared transaction is replayed against a new epoch.","triggerScenarios":"pre_commit stored epoch E1 in LanceDbPreCommitMetadata; the coordinator later calls commit_data with epoch E2 != E1, e.g. after a failover/restart, or a replayed commit for an epoch that was already committed/aborted.","commonSituations":"Recovery after a meta-node failure replaying old commit epochs; duplicated barrier delivery; retries of an already-committed epoch.","solutions":["Verify whether the epoch's data already landed in LanceDB before the error — if so, the stale prepared state can be safely discarded","Restart/recover the sink so pre_commit state is re-created at the current epoch","Ensure only one coordinator commits to the sink at a time (check for duplicate sink worker assignments)","If persistent after a single clean restart, report — it may indicate lost/incorrect pre-commit metadata"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// before committing, verify metadata epoch matches\nlet meta = LanceDbPreCommitMetadata::try_from_bytes(&commit_metadata)?;\nif meta.epoch != epoch { /* stale prepared txn: verify data landed, then discard */ }","typeGuard":"fn is_epoch_mismatch(err: &SinkError) -> bool {\n    matches!(err, SinkError::LanceDb(e) if e.to_string().contains(\"does not match commit epoch\"))\n}","tryCatchPattern":"match commit_result {\n    Err(e) if e.to_string().contains(\"does not match commit epoch\") => {\n        // stale prepared state from before recovery: verify data, then proceed with new epoch\n    }\n    other => other,\n}","preventionTips":["After a crash/recovery, verify whether the previous epoch's data landed in LanceDB before replaying","Ensure single-coordinator semantics for the sink (no duplicated barrier commits)","Check that pre-commit metadata persists and restores correctly across restarts"],"tags":["lancedb","sink","epoch-mismatch","two-phase-commit","recovery"],"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"}