{"record":{"id":"c6a25dc6d5da605c","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-merger-received-a-non-checkpoi","errorCode":null,"errorMessage":"iceberg pk-index merger {} received a non-checkpoint compaction resume barrier {:?}","messagePattern":"iceberg pk-index merger (.+?) received a non-checkpoint compaction resume barrier (.+?)","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/position_delete_merger.rs","lineNumber":130,"sourceCode":"                        debug_assert_eq!(op, risingwave_common::array::Op::Insert);\n                        let file_path = row\n                            .datum_at(0)\n                            .map(|d| d.into_utf8())\n                            .context(\"file_path should not be null\")?;\n                        let position = row\n                            .datum_at(1)\n                            .context(\"position should not be null\")?\n                            .into_int64();\n                        self.handler\n                            .write(file_path, position)\n                            .map_err(|e| StreamExecutorError::sink_error(e, self.sink_id))?;\n                    }\n                }\n                Message::Barrier(barrier) => {\n                    barrier.assume_no_update_vnode_bitmap(self.actor_id)?;\n                    let compaction_resumed = self.compaction_resume_task_id(&barrier);\n                    if compaction_resumed.is_some() && !barrier.is_checkpoint() {\n                        bail!(\n                            \"iceberg pk-index merger {} received a non-checkpoint compaction resume barrier {:?}\",\n                            self.sink_id,\n                            barrier.epoch\n                        );\n                    }\n\n                    let mut metadata = None;\n                    if barrier.is_checkpoint() {\n                        metadata = self\n                            .handler\n                            .flush()\n                            .await\n                            .map_err(|e| StreamExecutorError::sink_error(e, self.sink_id))?;\n                    }\n\n                    if let Some(metadata) = metadata\n                        && metadata.metadata.is_some()\n                    {","sourceCodeStart":112,"sourceCodeEnd":148,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/position_delete_merger.rs#L112-L148","documentation":"The iceberg pk-index merger only allows a barrier that resumes a paused compaction task (iceberg_pk_index_compaction resume) to arrive at a checkpoint barrier, because resume must align with a persistent epoch boundary. If a barrier carries a compaction resume task id but is not a checkpoint, execute_inner aborts with this error. Resuming on a non-checkpoint barrier would break recovery invariants.","triggerScenarios":"Raised in execute_inner when self.compaction_resume_task_id(&barrier) returns Some while barrier.is_checkpoint() is false — the meta node issued a compaction-resume mutation attached to a non-checkpoint barrier.","commonSituations":"Meta scheduler bug issuing resume mutations on high-frequency non-checkpoint barriers; misconfigured checkpoint interval making resumes attach to the wrong barrier type; version skew between meta and stream nodes.","solutions":["Check the compaction scheduler on the meta side to ensure resume mutations are only attached to checkpoint barriers.","Align checkpoint interval so compaction resumes can be delivered at checkpoints.","Verify meta/stream node version compatibility.","Restart the fragment; this is an invariant violation and recovery should re-plan the compaction task.","File a bug with the barrier details (epoch) if reproducible."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// caller-side (meta/scheduler) invariant check before sending resume\nif compaction_resume_pending && !barrier.is_checkpoint() {\n    // defer the resume mutation to the next checkpoint barrier\n}","typeGuard":"fn resume_is_legal(barrier: &Barrier, resume_task: Option<IcebergCompactionTaskId>) -> bool {\n    resume_task.is_none() || barrier.is_checkpoint()\n}","tryCatchPattern":"match res {\n    Err(e) if e.to_string().contains(\"non-checkpoint compaction resume barrier\") => {\n        error!(\"compaction resume delivered off-checkpoint; restarting fragment from checkpoint\");\n    }\n    other => other?,\n}","preventionTips":["Ensure the meta compaction scheduler only attaches resume mutations to checkpoint barriers.","Keep checkpoint intervals regular so resumes have a predictable boundary.","Run matching meta/stream versions."],"tags":["iceberg","compaction","barrier","checkpoint","streaming-executor"],"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"}