{"record":{"id":"5b6a15f235d4fc3e","repo":"risingwavelabs/risingwave","slug":"compaction-resolver-sink-expected-initial-begin","errorCode":null,"errorMessage":"compaction resolver sink {} expected initial begin barrier, got {:?}","messagePattern":"compaction resolver sink (.+?) expected initial begin barrier, got (.+?)","errorType":"validation","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs","lineNumber":99,"sourceCode":"    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,\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","sourceCodeStart":81,"sourceCodeEnd":117,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/compaction_resolver.rs#L81-L117","documentation":"`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.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Validate the incoming barrier before initializing the resolver:\nlet valid = barrier.is_checkpoint()\n    && barrier.context.as_ref().map_or(false, |c| c.phase == Phase::Begin as i32);","typeGuard":"fn is_begin_checkpoint(b: &StreamChunkBarrier) -> bool {\n    b.is_checkpoint()\n        && b.context.as_ref().is_some_and(|c| c.phase == Phase::Begin as i32)\n}","tryCatchPattern":"if let Err(e) = executor.execute_inner().await {\n    if e.to_string().contains(\"expected initial begin barrier\") {\n        // restart the actor from a checkpoint so barrier order restarts at Begin\n    }\n}","preventionTips":["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."],"tags":["stream-executor","iceberg","compaction-resolver","barrier","invalid-state"],"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"}