{"record":{"id":"7d61cd6db0ce5fea","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-writer-expected-matching","errorCode":null,"errorMessage":"iceberg pk-index writer {} expected matching {:?} context for task {}, got {:?}","messagePattern":"iceberg pk-index writer (.+?) expected matching (.+?) context for task (.+?), got (.+?)","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/writer.rs","lineNumber":332,"sourceCode":"    ) -> StreamExecutorResult<()> {\n        if !barrier.is_checkpoint() || barrier.epoch.prev != expected_prev {\n            bail!(\n                \"iceberg pk-index writer {} expected checkpoint {:?} starting at {}, got {:?}\",\n                self.sink_id,\n                expected_phase,\n                expected_prev,\n                barrier\n            );\n        }\n        match barrier.iceberg_pk_index_compaction() {\n            Some(context)\n                if context.sink_id == self.sink_id\n                    && context.task_id == expected_task\n                    && context.phase == expected_phase as i32 =>\n            {\n                Ok(())\n            }\n            _ => bail!(\n                \"iceberg pk-index writer {} expected matching {:?} context for task {}, got {:?}\",\n                self.sink_id,\n                expected_phase,\n                expected_task,\n                barrier\n            ),\n        }\n    }\n\n    fn validate_aligned_barriers(\n        &self,\n        left: &Barrier,\n        right: &Barrier,\n    ) -> StreamExecutorResult<()> {\n        if left.epoch != right.epoch || left.kind != right.kind || left.mutation != right.mutation {\n            bail!(\n                \"iceberg pk-index writer {} received mismatched left/right barriers: left={:?}, right={:?}\",\n                self.sink_id,","sourceCodeStart":314,"sourceCodeEnd":350,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/writer.rs#L314-L350","documentation":"When a compaction barrier arrives, the writer checks its embedded iceberg_pk_index_compaction context matches the current sink_id, task_id, and expected phase. This error means the context belongs to a different phase, task, or sink, so the writer cannot apply the barrier to its current compaction state machine.","triggerScenarios":"Raised in validate_compaction_barrier when the barrier's context fails the check context.sink_id == self.sink_id && context.task_id == expected_task && context.phase == expected_phase — e.g. a Begin barrier for a different task id, or an End/Apply barrier arriving while expecting Begin.","commonSituations":"Meta dispatched a compaction task to the wrong sink instance; stale compaction task ids after job restart/recovery; phase transitions racing so an old task's barrier arrives during a new task; actor migration reusing stale context.","solutions":["Log the barrier's context (sink_id, task_id, phase) and compare with expected values to identify the mismatch kind.","Restart from the last checkpoint to clear stale compaction task state.","Check meta's compaction task assignment for duplicate or stale task ids.","Ensure recovery doesn't reuse pre-restart compaction task ids.","Report to maintainers with the context dump if it recurs on healthy clusters."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// verify the barrier's compaction context before feeding it to the writer\nif let Some(ctx) = barrier.iceberg_pk_index_compaction() {\n    assert_eq!(ctx.sink_id, writer.sink_id);\n    assert_eq!(ctx.task_id, expected_task);\n}","typeGuard":"fn context_matches(barrier: &Barrier, sink_id: u32, task: IcebergCompactionTaskId, phase: Phase) -> bool {\n    matches!(barrier.iceberg_pk_index_compaction(),\n        Some(ctx) if ctx.sink_id == sink_id && ctx.task_id == task && ctx.phase == phase as i32)\n}","tryCatchPattern":"match writer.execute_resolving_right(msg).await {\n    Err(e) if e.to_string().contains(\"expected matching\") => {\n        error!(\"compaction context mismatch; restart fragment from checkpoint\");\n    }\n    other => other?,\n}","preventionTips":["Ensure meta assigns compaction tasks to exactly one writer per sink.","Clear stale task ids on recovery.","Test compaction resume/restart scenarios in CI."],"tags":["iceberg","compaction","barrier","phase","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"}