risingwavelabs/risingwave · error · StreamExecutorError

compaction resolver sink {} task {} expected end barrier, go

Error message

compaction resolver sink {} task {} expected end barrier, got {:?}

What it means

`validate_resolver_end_barrier` requires the concluding barrier for a resolver task to be a checkpoint End-phase barrier with matching sink id, task id, and no `resolver_task_input`. Any deviation produces this error including the actual barrier seen.

Source

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

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
                && context.task_id == task_id
                && context.phase == Phase::End as i32
                && context.resolver_task_input.is_none() =>
        {
            Ok(())
        }
        _ => Err(StreamExecutorError::from(anyhow!(
            "compaction resolver sink {} task {} expected end barrier, got {:?}",
            sink_id,
            task_id,
            barrier
        ))),
    }
}

impl CompactionResolverExecutor {
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        ctx: ActorContextRef,
        sink_id: SinkId,
        iceberg_config: IcebergConfig,
        pk_indices: Vec<usize>,
        pk_data_types: Vec<DataType>,
        chunk_size: usize,
        local_barrier_manager: LocalBarrierManager,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Confirm the compaction coordinator emits one correctly formed End checkpoint barrier per Begin, with matching sink and task ids.
2. Restart the affected sink/job to re-run the resolver cycle from a clean state.
3. Inspect the barrier debug output in the error message to identify which field (phase, task id, sink id, or task input) mismatched.

// Validate the End barrier before committing the resolver cycle:
let ok = barrier.is_checkpoint()
    && ctx.phase == Phase::End as i32
    && ctx.sink_id == sink_id
    && ctx.task_id == task_id
    && ctx.resolver_task_input.is_none();

fn is_valid_end_barrier(b: &StreamChunkBarrier, sink_id: u64, task_id: u64) -> bool {
    b.is_checkpoint()
        && b.context.as_ref().is_some_and(|c| {
            c.sink_id == sink_id
                && c.task_id == task_id
                && c.phase == Phase::End as i32
                && c.resolver_task_input.is_none()
        })
}

if let Err(e) = validate_resolver_end_barrier(&barrier, sink_id, task_id) {
    // log the barrier Debug output embedded in the error, then restart the resolver cycle
}

When it happens

Trigger: During a compaction resolver cycle's end: the End barrier is non-checkpoint, has the wrong phase/task id/sink id, or unexpectedly carries a `resolver_task_input` at src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs:124.

Common situations: Concurrent compaction tasks interleaving barriers so a mismatched task id arrives; recovery replaying stale barriers; coordinator bugs emitting End barriers with residual task input.

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

• compaction resolver sink {} task {} missing resolver task in
• compaction resolver sink {} expected initial begin barrier,
• `warehouse.path` must be set
• Unsupported scheme: {}
• `catalog.type` must not be set when `hosted_catalog` is set

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