{"record":{"id":"3c2d65a17ebf802c","repo":"risingwavelabs/risingwave","slug":"compaction-resolver-sink-task-expected-end-b","errorCode":null,"errorMessage":"compaction resolver sink {} task {} expected end barrier, got {:?}","messagePattern":"compaction resolver sink (.+?) task (.+?) expected end barrier, got (.+?)","errorType":"validation","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":124,"sourceCode":"\nfn validate_resolver_end_barrier(\n    sink_id: SinkId,\n    barrier: &Barrier,\n    begin: &Barrier,\n    task_id: IcebergCompactionTaskId,\n) -> StreamExecutorResult<()> {\n    match barrier.iceberg_pk_index_compaction() {\n        Some(context)\n            if barrier.is_checkpoint()\n                && barrier.epoch.prev == begin.epoch.curr\n                && context.sink_id == sink_id\n                && context.task_id == task_id\n                && context.phase == Phase::End as i32\n                && context.resolver_task_input.is_none() =>\n        {\n            Ok(())\n        }\n        _ => Err(StreamExecutorError::from(anyhow!(\n            \"compaction resolver sink {} task {} expected end barrier, got {:?}\",\n            sink_id,\n            task_id,\n            barrier\n        ))),\n    }\n}\n\nimpl CompactionResolverExecutor {\n    #[expect(clippy::too_many_arguments)]\n    pub fn new(\n        ctx: ActorContextRef,\n        sink_id: SinkId,\n        iceberg_config: IcebergConfig,\n        pk_indices: Vec<usize>,\n        pk_data_types: Vec<DataType>,\n        chunk_size: usize,\n        local_barrier_manager: LocalBarrierManager,","sourceCodeStart":106,"sourceCodeEnd":142,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L106-L142","documentation":"`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.","triggerScenarios":"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.","commonSituations":"Concurrent compaction tasks interleaving barriers so a mismatched task id arrives; recovery replaying stale barriers; coordinator bugs emitting End barriers with residual task input.","solutions":["Confirm the compaction coordinator emits one correctly formed End checkpoint barrier per Begin, with matching sink and task ids.","Restart the affected sink/job to re-run the resolver cycle from a clean state.","Inspect the barrier debug output in the error message to identify which field (phase, task id, sink id, or task input) mismatched."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Validate the End barrier before committing the resolver cycle:\nlet ok = barrier.is_checkpoint()\n    && ctx.phase == Phase::End as i32\n    && ctx.sink_id == sink_id\n    && ctx.task_id == task_id\n    && ctx.resolver_task_input.is_none();","typeGuard":"fn is_valid_end_barrier(b: &StreamChunkBarrier, sink_id: u64, task_id: u64) -> bool {\n    b.is_checkpoint()\n        && b.context.as_ref().is_some_and(|c| {\n            c.sink_id == sink_id\n                && c.task_id == task_id\n                && c.phase == Phase::End as i32\n                && c.resolver_task_input.is_none()\n        })\n}","tryCatchPattern":"if let Err(e) = validate_resolver_end_barrier(&barrier, sink_id, task_id) {\n    // log the barrier Debug output embedded in the error, then restart the resolver cycle\n}","preventionTips":["Ensure the coordinator emits exactly one End checkpoint barrier per Begin with matching ids.","Avoid interleaving multiple compaction tasks that can cross-deliver End barriers.","Inspect the {:?} barrier dump in the error to pinpoint the mismatched field."],"tags":["stream-executor","iceberg","compaction-resolver","barrier","validation"],"backgroundTag":"invalid-state-transition","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"}