What causes "compaction resolver sink {} expected initial begin barrier, got {:?}" in risingwavelabs/risingwave?

Restart the affected stream job/actor so recovery begins from a checkpoint and the first barrier is a Begin checkpoint.

risingwavelabs/risingwave · error · StreamExecutorError

compaction resolver sink {} expected initial begin barrier,

Error message

compaction resolver sink {} expected initial begin barrier, got {:?}

What it means

`resolver_task_from_initial_barrier` expects the first barrier it inspects for a sink to be a checkpoint Begin-phase barrier. Any other barrier (non-checkpoint, wrong phase, or no matching context) yields this error reporting the sink id and the unexpected barrier.

Source

Thrown at src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs:99

    sink_id: SinkId,
    barrier: &Barrier,
) -> StreamExecutorResult<(IcebergCompactionTaskId, ResolverTaskInput)> {
    match barrier.iceberg_pk_index_compaction() {
        Some(context)
            if barrier.is_checkpoint()
                && context.sink_id == sink_id
                && context.phase == Phase::Begin as i32 =>
        {
            let task_input = context.resolver_task_input.clone().ok_or_else(|| {
                StreamExecutorError::from(anyhow!(
                    "compaction resolver sink {} task {} missing resolver task input",
                    sink_id,
                    context.task_id
                ))
            })?;
            Ok((context.task_id, task_input))
        }
        _ => Err(StreamExecutorError::from(anyhow!(
            "compaction resolver sink {} expected initial begin barrier, got {:?}",
            sink_id,
            barrier
        ))),
    }
}

fn validate_resolver_end_barrier(
    sink_id: SinkId,
    barrier: &Barrier,
    begin: &Barrier,
    task_id: IcebergCompactionTaskId,
) -> StreamExecutorResult<()> {
    match barrier.iceberg_pk_index_compaction() {
        Some(context)
            if barrier.is_checkpoint()
                && barrier.epoch.prev == begin.epoch.curr
                && context.sink_id == sink_id

View on GitHub (pinned to 6469eb736d)

Solutions

Restart the affected stream job/actor so recovery begins from a checkpoint and the first barrier is a Begin checkpoint.
Check the compaction coordinator logs for out-of-order or missing Begin barriers.
Upgrade to a version with consistent barrier phase ordering if mixed-version deployment is suspected.

Defensive patterns

Strategy: retry

Validate before calling

// Validate the incoming barrier before initializing the resolver:
let valid = barrier.is_checkpoint()
    && barrier.context.as_ref().map_or(false, |c| c.phase == Phase::Begin as i32);

Type guard

fn is_begin_checkpoint(b: &StreamChunkBarrier) -> bool {
    b.is_checkpoint()
        && b.context.as_ref().is_some_and(|c| c.phase == Phase::Begin as i32)
}

Try / catch

if let Err(e) = executor.execute_inner().await {
    if e.to_string().contains("expected initial begin barrier") {
        // restart the actor from a checkpoint so barrier order restarts at Begin
    }
}

Prevention

Recover actors only from checkpoints so the first barrier seen is a Begin checkpoint.
Keep coordinator and executor versions consistent to avoid phase-order mismatches.
Alert on compaction phase transitions that skip Begin.

When it happens

Trigger: Initializing the compaction resolver when the first barrier delivered is not a checkpoint `Begin` barrier — e.g. actor starts mid-phase after recovery, receives an `End` barrier first, or a non-checkpoint barrier arrives first at src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs:99.

Common situations: Failover/recovery starting between phases so the actor's first barrier is out of order; barrier misrouting to the wrong actor; protocol bugs in the compaction coordinator emitting phases in the wrong order.

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

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