{"record":{"id":"a8e2a85db2b5a05d","repo":"risingwavelabs/risingwave","slug":"new-item-epoch-does-not-match-current-chunk-off","errorCode":null,"errorMessage":"new item epoch {} does not match current chunk offset epoch {}","messagePattern":"new item epoch (.+?) does not match current chunk offset epoch (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/log_store.rs","lineNumber":96,"sourceCode":"        if *self >= next_offset {\n            bail!(\n                \"next offset {:?} should be later than current offset {:?}\",\n                next_offset,\n                self\n            )\n        } else {\n            Ok(())\n        }\n    }\n\n    pub fn check_next_item_epoch(&self, epoch: u64) -> LogStoreResult<()> {\n        match self {\n            TruncateOffset::Chunk {\n                epoch: offset_epoch,\n                ..\n            } => {\n                if epoch != *offset_epoch {\n                    bail!(\n                        \"new item epoch {} does not match current chunk offset epoch {}\",\n                        epoch,\n                        offset_epoch\n                    );\n                }\n            }\n            TruncateOffset::Barrier {\n                epoch: offset_epoch,\n            } => {\n                if epoch <= *offset_epoch {\n                    bail!(\n                        \"new item epoch {} does not exceed barrier offset epoch {}\",\n                        epoch,\n                        offset_epoch\n                    );\n                }\n            }\n        }","sourceCodeStart":78,"sourceCodeEnd":114,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/log_store.rs#L78-L114","documentation":"`check_next_item_epoch` on a `TruncateOffset::Chunk` requires that any new item written to the log store carries exactly the same epoch as the current chunk offset epoch. A mismatch means an item from a different epoch is being appended into a chunk opened for another epoch.","triggerScenarios":"Calling `check_next_item_epoch(epoch)` while the current offset is `TruncateOffset::Chunk { epoch: offset_epoch, .. }` and `epoch != offset_epoch`; typically when a sink item's epoch changed mid-chunk without closing the chunk first.","commonSituations":"Sink writers mixing items from adjacent epochs; barrier handling that fails to flush/rotate the chunk before the epoch advances.","solutions":["Close/rotate the current chunk and advance the offset when the item's epoch differs from the chunk epoch","Verify the item's epoch is stamped from the same source as the chunk epoch","Check barrier handling so epoch transitions always precede new items"],"exampleFix":"// before\nstore.check_next_item_epoch(new_epoch); // chunk opened with old_epoch\n// after\nif new_epoch != chunk_epoch {\n    store.flush_chunk_and_advance(new_epoch);\n}\nstore.check_next_item_epoch(new_epoch);","handlingStrategy":"validation","validationCode":"assert_eq!(item_epoch, chunk_epoch, \"item epoch must match open chunk epoch\");","typeGuard":"fn matches_chunk_epoch(epoch: u64, offset: &TruncateOffset) -> bool {\n    matches!(offset, TruncateOffset::Chunk { epoch: e, .. } if *e == epoch)\n}","tryCatchPattern":"match offset.check_next_item_epoch(epoch) {\n    Err(e) => { store.rotate_chunk(epoch)?; store.check_next_item_epoch(epoch)?; }\n    Ok(()) => {}\n}","preventionTips":["Rotate the chunk whenever the item epoch changes","Derive item epochs and chunk epochs from the same barrier stream","Test epoch-transition paths in sink writers"],"tags":["rust","streaming","epoch-mismatch"],"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"}