{"record":{"id":"fb649a4c1d665bb0","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-writer-received-unexpected-end","errorCode":null,"errorMessage":"iceberg pk-index writer {} received unexpected End in Normal mode for task {}","messagePattern":"iceberg pk-index writer (.+?) received unexpected End in Normal mode for task (.+?)","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/writer.rs","lineNumber":367,"sourceCode":"                \"iceberg pk-index writer {} received mismatched left/right barriers: left={:?}, right={:?}\",\n                self.sink_id,\n                left,\n                right\n            );\n        }\n        Ok(())\n    }\n\n    fn compaction_begin(\n        &self,\n        barrier: &Barrier,\n    ) -> StreamExecutorResult<Option<IcebergCompactionTaskId>> {\n        let context = match barrier.iceberg_pk_index_compaction() {\n            Some(context) if context.sink_id == self.sink_id => context,\n            _ => return Ok(None),\n        };\n        if context.phase == Phase::End as i32 {\n            bail!(\n                \"iceberg pk-index writer {} received unexpected End in Normal mode for task {}\",\n                self.sink_id,\n                context.task_id\n            );\n        }\n        if context.phase != Phase::Begin as i32 {\n            bail!(\n                \"iceberg pk-index writer {} expected Begin context for task {}, got {:?}\",\n                self.sink_id,\n                context.task_id,\n                context.phase\n            );\n        }\n        Ok(Some(context.task_id))\n    }\n\n    #[try_stream(ok = Message, error = StreamExecutorError)]\n    async fn execute_inner(mut self) {","sourceCodeStart":349,"sourceCodeEnd":385,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/writer.rs#L349-L385","documentation":"In Normal mode the writer handles a compaction Begin barrier to enter the compaction flow, but a Phase::End context must be consumed while already in compaction mode, not Normal mode. Receiving an End while Normal means the writer missed the corresponding Begin (or already returned to Normal prematurely), so the compaction lifecycle is inconsistent.","triggerScenarios":"Raised in compaction_begin (called from execute_normal) when barrier.iceberg_pk_index_compaction() yields a context for this sink whose phase is Phase::End — i.e. the compaction-terminating barrier arrives while the writer is still in Normal mode for that task.","commonSituations":"The Begin barrier was lost or consumed incorrectly during recovery/replay; meta sent End without a preceding Begin after a task restart; version skew where an older writer misses the Begin-phase barrier type.","solutions":["Restart the fragment from the last checkpoint to restore a consistent compaction lifecycle.","Check meta's compaction scheduler for an End issued without a matching Begin.","Inspect logs for the missing Begin barrier (task_id in the error identifies the task).","Verify meta/stream node version compatibility.","Report with the task id if the scheduler reproducibly skips Begin."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// reject End contexts when the writer is in Normal mode, before dispatch\nif mode == Mode::Normal && ctx.phase == Phase::End as i32 {\n    // ignore or log; do not feed to compaction_begin\n}","typeGuard":"fn begin_allowed(ctx: &IcebergPkIndexCompactionContext) -> bool {\n    ctx.phase == Phase::Begin as i32\n}","tryCatchPattern":"match res {\n    Err(e) if e.to_string().contains(\"unexpected End in Normal mode\") => {\n        error!(\"compaction lifecycle desync (missed Begin); restart from checkpoint\");\n    }\n    other => other?,\n}","preventionTips":["Guarantee Begin/End phase pairs are delivered in order and never split across recoveries.","Log every compaction phase barrier per task_id for auditing.","Pin matching meta and stream versions."],"tags":["iceberg","compaction","barrier","phase","state-machine"],"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"}