{"record":{"id":"55fcce7f1d04fc08","repo":"risingwavelabs/risingwave","slug":"iceberg-pk-index-writer-received-mismatched-lef","errorCode":null,"errorMessage":"iceberg pk-index writer {} received mismatched left/right barriers: left={:?}, right={:?}","messagePattern":"iceberg pk-index writer (.+?) received mismatched left/right barriers: left=(.+?), right=(.+?)","errorType":"exception","errorClass":"StreamExecutorError","httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/iceberg_with_pk_index/writer.rs","lineNumber":348,"sourceCode":"                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,\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 {","sourceCodeStart":330,"sourceCodeEnd":366,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/iceberg_with_pk_index/writer.rs#L330-L366","documentation":"The pk-index writer is a two-input executor whose left (data/sink) and right (resolver) streams must see identical barriers: same epoch, same kind, and same mutation. validate_aligned_barriers enforces this; a mismatch means the two streams have diverged and continuing could corrupt the pk index or the Iceberg commit.","triggerScenarios":"Raised in validate_aligned_barriers (called from execute_inner, execute_normal, execute_aligning_replacement_input) when left.epoch != right.epoch, left.kind != right.kind, or left.mutation != right.mutation for the pair of barriers drawn from the two inputs.","commonSituations":"Uneven upstream backpressure or failure causing one input to advance epochs ahead of the other; a mutation (e.g. pause/resume, config change) applied to only one branch; topology rewiring after scale-in/out breaking barrier symmetric delivery.","solutions":["Log both barriers and identify which field diverged (epoch, kind, or mutation).","Restart the fragment from the last checkpoint so both inputs realign.","Check the fragment's actor graph for asymmetric mutation application (one side only).","Verify barrier dispatch from meta reaches both inputs symmetrically.","File a bug with both barrier Debug dumps if reproducible."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// assert symmetry before forwarding barriers to the two-input writer\nassert_eq!(left.epoch, right.epoch);\nassert_eq!(left.kind, right.kind);\nassert_eq!(left.mutation, right.mutation);","typeGuard":"fn barriers_aligned(left: &Barrier, right: &Barrier) -> bool {\n    left.epoch == right.epoch && left.kind == right.kind && left.mutation == right.mutation\n}","tryCatchPattern":"if let Err(e) = writer.execute_inner(...).await {\n    if e.to_string().contains(\"mismatched left/right barriers\") {\n        error!(\"input streams diverged; recovering both inputs from last checkpoint\");\n    }\n    return Err(e);\n}","preventionTips":["Keep the fragment graph symmetric so mutations apply to both branches.","Monitor per-input epoch progress and alert on skew.","Recover both inputs from the same checkpoint after failover."],"tags":["iceberg","barrier","two-input-executor","alignment","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"}