risingwavelabs/risingwave · error

next offset {:?} should be later than current offset {:?}

Error message

next offset {:?} should be later than current offset {:?}

What it means

LogStore ordering invariant check: `check_next_offset` on the current truncate offset rejects any proposed next truncate offset that is not strictly later than the current one. It guards the sink log store against rewinding or replaying offsets, which would corrupt the sink's at-least-once/exactly-once delivery bookkeeping.

Source

Thrown at src/connector/src/sink/log_store.rs:79

}

impl TruncateOffset {
    pub fn next_chunk_id(&self) -> ChunkId {
        match self {
            TruncateOffset::Chunk { chunk_id, .. } => chunk_id + 1,
            TruncateOffset::Barrier { .. } => 0,
        }
    }

    pub fn epoch(&self) -> u64 {
        match self {
            TruncateOffset::Chunk { epoch, .. } | TruncateOffset::Barrier { epoch } => *epoch,
        }
    }

    pub fn check_next_offset(&self, next_offset: TruncateOffset) -> LogStoreResult<()> {
        if *self >= next_offset {
            bail!(
                "next offset {:?} should be later than current offset {:?}",
                next_offset,
                self
            )
        } else {
            Ok(())
        }
    }

    pub fn check_next_item_epoch(&self, epoch: u64) -> LogStoreResult<()> {
        match self {
            TruncateOffset::Chunk {
                epoch: offset_epoch,
                ..
            } => {
                if epoch != *offset_epoch {
                    bail!(
                        "new item epoch {} does not match current chunk offset epoch {}",

View on GitHub (pinned to 6469eb736d)

Solutions

1. Fix the sink writer to only advance the truncate offset (next epoch must be strictly greater)
2. Check for stale offsets cached across barrier/recovery and refresh them before calling check_next_offset
3. Ensure only one writer drives the log store for a given sink

Example fix

// before
store.check_next_offset(TruncateOffset::Barrier { epoch: old_epoch });
// after
if epoch > current_epoch {
    store.check_next_offset(TruncateOffset::Barrier { epoch });
}

fn can_advance(current: &TruncateOffset, next: &TruncateOffset) -> bool {
    next.epoch_value() > current.epoch_value()
}

fn is_later(next: &TruncateOffset, current: &TruncateOffset) -> bool {
    match (next, current) {
        (TruncateOffset::Chunk { epoch: n, .. } | TruncateOffset::Barrier { epoch: n },
         TruncateOffset::Chunk { epoch: c, .. } | TruncateOffset::Barrier { epoch: c }) => n > c,
    }
}

match store.check_next_offset(next) {
    Ok(()) => store.advance(next),
    Err(e) => { warn!("stale offset skipped: {e}"); /* refresh offset from barrier */ }
}

When it happens

Trigger: Calling `LogStore::check_next_offset(next_offset)` with a `TruncateOffset` whose epoch is <= the currently stored offset's epoch, e.g. a sink writer producing stale chunk/barrier offsets after a failure or epoch regression.

Common situations: A sink implementation truncating with an offset captured before a barrier; epoch values reused after recovery; a sink connector that reorders epochs under concurrent writers.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• new item epoch {} does not match current chunk offset epoch
• new item epoch {} does not exceed barrier offset epoch {}
• Division by zero
• Array error: {0}
• file_scan function is not supported in streaming mode

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/bf05c6283ccf4e3a. Report an issue: GitHub.