{"record":{"id":"eff855726348429c","repo":"risingwavelabs/risingwave","slug":"parallelizedcdcbackfillexecutor-expects-either-mut","errorCode":null,"errorMessage":"ParallelizedCdcBackfillExecutor expects either Mutation::Add or Mutation::Update to initialize CDC table snapshot splits.","messagePattern":"ParallelizedCdcBackfillExecutor expects either Mutation::Add or Mutation::Update to initialize CDC table snapshot splits\\.","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/stream/src/executor/backfill/cdc/cdc_backill_v2.rs","lineNumber":167,"sourceCode":"            handle_toast_columns,\n        )\n        .boxed();\n        let mut next_reset_barrier = Some(first_barrier);\n        let mut is_reset = false;\n        let mut state_impl = ParallelizedCdcBackfillState::new(self.state_table);\n        // The buffered chunks have already been mapped.\n        let mut upstream_chunk_buffer: Vec<StreamChunk> = vec![];\n\n        // Need reset on CDC table snapshot splits reschedule.\n        'with_cdc_table_snapshot_splits: loop {\n            assert!(upstream_chunk_buffer.is_empty());\n            let reset_barrier = next_reset_barrier.take().unwrap();\n            let all_snapshot_splits = match reset_barrier.mutation.as_deref() {\n                Some(Mutation::Add(add)) => &add.actor_cdc_table_snapshot_splits.splits,\n\n                Some(Mutation::Update(update)) => &update.actor_cdc_table_snapshot_splits.splits,\n                _ => {\n                    return Err(anyhow::anyhow!(\"ParallelizedCdcBackfillExecutor expects either Mutation::Add or Mutation::Update to initialize CDC table snapshot splits.\").into());\n                }\n            };\n            let mut actor_snapshot_splits = vec![];\n            let mut generation = None;\n            // TODO(zw): optimization: remove consumed splits to reduce barrier size for downstream.\n            if let Some((splits, snapshot_generation)) = all_snapshot_splits.get(&self.actor_ctx.id)\n            {\n                actor_snapshot_splits = splits\n                    .iter()\n                    .map(|s: &CdcTableSnapshotSplitRaw| {\n                        let de = RowDeserializer::new(\n                            cdc_table_snapshot_split_column\n                                .iter()\n                                .map(Field::data_type)\n                                .collect_vec(),\n                        );\n                        let left_bound_inclusive =\n                            de.deserialize(s.left_bound_inclusive.as_ref()).unwrap();","sourceCodeStart":149,"sourceCodeEnd":185,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/executor/backfill/cdc/cdc_backill_v2.rs#L149-L185","documentation":"`ParallelizedCdcBackfillExecutor::execute_inner` expects the incoming reset barrier's mutation to be `Mutation::Add` or `Mutation::Update`, both of which carry CDC table snapshot splits used to initialize backfill. Any other mutation kind (or none) on a reset barrier is invalid initialization input, so it returns this error.","triggerScenarios":"A reset/merge barrier reaches the CDC backfill executor without the expected Add/Update mutation carrying `actor_cdc_table_snapshot_splits` — e.g. reschedule/merge logic sends a wrong mutation type, or the barrier was constructed without snapshot splits.","commonSituations":"Meta-node rescheduling bugs during CDC table backfill, recovery scenarios where the reset barrier is replayed with a different mutation, or version-skew between meta and compute nodes.","solutions":["Check meta logs around the CDC table creation/reschedule time to find which mutation was sent to this actor.","Retry the CDC table creation (drop and re-create the CDC table / CDC source) so a fresh, well-formed reset barrier is issued.","Ensure meta and compute node versions match; version skew can change mutation payload shapes.","If reproducible on one version, file with the barrier type observed; upgrade in case the rescheduler fix already landed."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Wrap CDC table creation/retry loop:\nmatch create_cdc_table_with_backfill().await {\n    Err(e) if e.to_string().contains(\"expects either Mutation::Add or Mutation::Update\") => {\n        log::warn!(\"malformed reset barrier during CDC backfill; retrying creation\");\n        drop_cdc_table().await?;\n        create_cdc_table_with_backfill().await?;\n    }\n    other => other?,\n}","preventionTips":["Keep meta and compute nodes on identical versions so mutation payloads match.","Avoid rescheduling CDC backfill actors under heavy DDL load; check meta logs for reschedule storms.","Retry CDC table creation cleanly (drop then re-create) instead of forcing recovery."],"tags":["streaming","cdc","backfill","barrier","mutation"],"backgroundTag":"unexpected-response-shape","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"}