{"record":{"id":"ee8fe32a7f6a3eab","repo":"risingwavelabs/risingwave","slug":"compaction-resolver-sink-task-missing-resolv","errorCode":null,"errorMessage":"compaction resolver sink {} task {} missing resolver task input","messagePattern":"compaction resolver sink (.+?) task (.+?) missing resolver task input","errorType":"validation","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":91,"sourceCode":"    pk_data_types: Vec<DataType>,\n    chunk_size: usize,\n    local_barrier_manager: LocalBarrierManager,\n    barrier_receiver: UnboundedReceiver<Barrier>,\n    meta_client: MetaClient,\n}\n\nfn resolver_task_from_initial_barrier(\n    sink_id: SinkId,\n    barrier: &Barrier,\n) -> StreamExecutorResult<(IcebergCompactionTaskId, ResolverTaskInput)> {\n    match barrier.iceberg_pk_index_compaction() {\n        Some(context)\n            if barrier.is_checkpoint()\n                && context.sink_id == sink_id\n                && context.phase == Phase::Begin as i32 =>\n        {\n            let task_input = context.resolver_task_input.clone().ok_or_else(|| {\n                StreamExecutorError::from(anyhow!(\n                    \"compaction resolver sink {} task {} missing resolver task input\",\n                    sink_id,\n                    context.task_id\n                ))\n            })?;\n            Ok((context.task_id, task_input))\n        }\n        _ => Err(StreamExecutorError::from(anyhow!(\n            \"compaction resolver sink {} expected initial begin barrier, got {:?}\",\n            sink_id,\n            barrier\n        ))),\n    }\n}\n\nfn validate_resolver_end_barrier(\n    sink_id: SinkId,\n    barrier: &Barrier,","sourceCodeStart":73,"sourceCodeEnd":109,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L73-L109","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Ensure all compute/meta nodes run a version that serializes `resolver_task_input` on Begin barriers (avoid mixed-version recovery).","Restart the affected sink/job so a fresh Begin checkpoint barrier with the task input is produced.","Verify the barrier's `sink_id` matches the sink being recovered — a mismatch falls through and can surface a different error."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Before starting recovery, check the barrier context carries the payload:\nlet ok = context.sink_id == sink_id\n    && context.phase == Phase::Begin as i32\n    && context.resolver_task_input.is_some();","typeGuard":"fn has_task_input(ctx: &SinkContext) -> Option<&ResolverTaskInput> {\n    ctx.resolver_task_input.as_ref()\n}","tryCatchPattern":"match executor.run().await {\n    Err(e) if e.to_string().contains(\"missing resolver task input\") => {\n        // restart the job so a fresh Begin checkpoint barrier with the payload is produced\n    }\n    other => other?,\n}","preventionTips":["Keep all nodes on the same RisingWave version across recovery/upgrades.","Restart the affected sink cleanly rather than resuming across mixed-version checkpoints.","Monitor compaction-coordinator logs for barriers emitted without task input."],"tags":["stream-executor","iceberg","compaction-resolver","barrier","recovery"],"backgroundTag":"missing-required-argument","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}