risingwavelabs/risingwave · error · StreamExecutorError

compaction resolver sink {} task {} missing resolver task in

Error message

compaction resolver sink {} task {} missing resolver task input

What it means

The Iceberg compaction-resolver executor requires that the initial `Begin`-phase checkpoint barrier for a sink carries the serialized resolver task input (`resolver_task_input`). When a matching Begin barrier arrives without it, the executor returns this `StreamExecutorError` because the resolver task cannot be reconstructed.

Source

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

    pk_data_types: Vec<DataType>,
    chunk_size: usize,
    local_barrier_manager: LocalBarrierManager,
    barrier_receiver: UnboundedReceiver<Barrier>,
    meta_client: MetaClient,
}

fn resolver_task_from_initial_barrier(
    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,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Ensure all compute/meta nodes run a version that serializes `resolver_task_input` on Begin barriers (avoid mixed-version recovery).
2. Restart the affected sink/job so a fresh Begin checkpoint barrier with the task input is produced.
3. Verify the barrier's `sink_id` matches the sink being recovered — a mismatch falls through and can surface a different error.

// Before starting recovery, check the barrier context carries the payload:
let ok = context.sink_id == sink_id
    && context.phase == Phase::Begin as i32
    && context.resolver_task_input.is_some();

fn has_task_input(ctx: &SinkContext) -> Option<&ResolverTaskInput> {
    ctx.resolver_task_input.as_ref()
}

match executor.run().await {
    Err(e) if e.to_string().contains("missing resolver task input") => {
        // restart the job so a fresh Begin checkpoint barrier with the payload is produced
    }
    other => other?,
}

When it happens

Trigger: Recovering/resuming an iceberg_with_pk_index sink where the Begin-phase barrier's `SinkContext` lacks `resolver_task_input` (e.g. barrier produced by an older binary, stale recovery without the payload, or corrupted barrier context) at src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs:91.

Common situations: Cluster upgrade/rollback mixing versions where the barrier payload schema changed; recovery from a checkpoint where the context was dropped; sink ID mismatch causing the wrong context branch to be matched.

Understand the failure class

Background: "missing required argument" and "the following required arguments were not provided": what required-argument errors mean and how to fix them — this error's family across 20 libraries.

Related errors

• compaction resolver sink {} expected initial begin barrier,
• compaction resolver sink {} task {} expected end barrier, go
• since_timestamp requires at least one upstream table
• The cluster is recovering
• The cluster is recovering-adhoc

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