risingwavelabs/risingwave · error

epoch {} does not match with current epoch {}

Error message

epoch {} does not match with current epoch {}

What it means

This error is thrown by the KV log store's epoch validation (`validate_epoch`) when a barrier/write is issued for an epoch that differs from the epoch the stream state machine is currently tracking (`AllConsumingRow` or `BarrierAligning`). It means the log store expected all work for the current epoch to be finished and a matching-epoch operation before emitting a barrier, but got a different epoch. It protects the invariant that barrier epochs strictly follow the epochs of the data rows they cover.

Source

Thrown at src/stream/src/common/log_store_impl/kv_log_store/serde.rs:946

        self.row_streams.push(stream.into_future());
        while let Some((stream_epoch, _)) = self.not_started_streams.last()
            && *stream_epoch == epoch
        {
            let (_, stream) = self.not_started_streams.pop().expect("should not be empty");
            self.row_streams.push(stream.into_future());
        }
        self.stream_state = StreamState::AllConsumingRow { curr_epoch: epoch };
        Ok(true)
    }

    fn may_init_epoch(&mut self, epoch: u64) -> LogStoreResult<()> {
        let prev_epoch = match &self.stream_state {
            StreamState::Uninitialized => unreachable!("should have init"),
            StreamState::BarrierEmitted { prev_epoch } => *prev_epoch,
            StreamState::AllConsumingRow { curr_epoch }
            | StreamState::BarrierAligning { curr_epoch, .. } => {
                return if *curr_epoch != epoch {
                    Err(anyhow!(
                        "epoch {} does not match with current epoch {}",
                        epoch,
                        curr_epoch
                    ))
                } else {
                    Ok(())
                };
            }
        };

        if prev_epoch >= epoch {
            return Err(anyhow!(
                "epoch {} should be greater than prev epoch {}",
                epoch,
                prev_epoch
            ));
        }

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check that all actors have finished consuming rows of the current epoch before the barrier for the next epoch is propagated (barrier alignment).
2. Verify epoch assignment source (meta/Hummock epoch) has not gone backwards; log both epochs at the call site.
3. If this occurs during recovery, ensure the log store state was correctly restored from the manifest before replaying epochs.
4. Inspect upstream barrier injection for duplicated or out-of-order `Epoch` values in `Barrier` messages.

Example fix

// before: emitting barrier with a fresh epoch while rows of old epoch are still aligning
log_store.append_barrier(barrier.with_epoch(new_epoch)).await?;
// after: only advance epoch after the stream state confirms all rows of current epoch are consumed
if matches!(state, StreamState::AllConsumingRow { .. }) {
    log_store.append_barrier(barrier.with_epoch(new_epoch)).await?;
}

// Rust: verify epoch matches current stream state before appending a barrier
fn epoch_matches_state(state: &StreamState, epoch: u64) -> bool {
    match state {
        StreamState::AllConsumingRow { curr_epoch }
        | StreamState::BarrierAligning { curr_epoch, .. } => *curr_epoch == epoch,
        StreamState::BarrierEmitted { .. } | StreamState::Uninitialized => true,
    }
}
if !epoch_matches_state(&state, epoch) { /* defer or reject barrier */ }

fn is_row_phase(state: &StreamState) -> bool {
    matches!(state, StreamState::AllConsumingRow { .. } | StreamState::BarrierAligning { .. })
}

When it happens

Trigger: Calling the log store's epoch validation path (e.g. when appending a barrier or updating stream state) while `self.stream_state` is `StreamState::AllConsumingRow { curr_epoch }` or `StreamState::BarrierAligning { curr_epoch, .. }` and the incoming `epoch != curr_epoch`.

Common situations: Bugs in stream actor epoch progression (actors consuming rows of epoch N but a barrier for epoch N+1 or N-1 arriving first), recovery/bootstrap replaying epochs out of order, or mixing state from a restarted Hummock epoch assignment with stale in-memory stream state.

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

• epoch {} should be greater than prev epoch {}
• current epoch {} has exceeded the epoch {} of the stream tha
• Exchange executor should not have children!
• missing lagging barriers for direct log-store start from sna
• get different response epoch to commit epoch: {} {}

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