{"record":{"id":"a1a3fb8b9ccd4e7b","repo":"risingwavelabs/risingwave","slug":"failed-to-ack-aligned-initial-epoch-for-handl","errorCode":null,"errorMessage":"failed to ack aligned initial epoch {:?} for handle {}","messagePattern":"failed to ack aligned initial epoch (.+?) for handle (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":310,"sourceCode":"                .get_mut(&handle_id)\n                .ok_or_else(|| anyhow!(\"failed to find handle {} to start\", handle_id,))?;\n            handle.start(log_store_rewind_start_epoch).map_err(|_| {\n                anyhow!(\n                    \"failed to start {:?} for handle {}\",\n                    log_store_rewind_start_epoch,\n                    handle_id\n                )\n            })?;\n        }\n        Ok(())\n    }\n\n    fn ack_aligned_initial_epoch(&mut self, aligned_initial_epoch: u64) -> anyhow::Result<()> {\n        for (handle_id, handle) in &mut self.writer_handles {\n            handle\n                .ack_aligned_initial_epoch(aligned_initial_epoch)\n                .map_err(|_| {\n                    anyhow!(\n                        \"failed to ack aligned initial epoch {:?} for handle {}\",\n                        aligned_initial_epoch,\n                        handle_id\n                    )\n                })?;\n        }\n        Ok(())\n    }\n\n    fn ack_commit(\n        &mut self,\n        epoch: u64,\n        handle_ids: impl IntoIterator<Item = HandleId>,\n    ) -> anyhow::Result<()> {\n        for handle_id in handle_ids {\n            let handle = self.writer_handles.get_mut(&handle_id).ok_or_else(|| {\n                anyhow!(\n                    \"failed to find handle {} when acknowledging the commit for epoch {}\",","sourceCodeStart":292,"sourceCodeEnd":328,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L292-L328","documentation":"In `ack_aligned_initial_epoch`, every registered writer handle must acknowledge the aligned initial epoch; if any handle's `ack_aligned_initial_epoch` returns `Err`, this wrapper error is raised. It indicates the handle's writer side cannot accept the aligned epoch — usually because the writer actor has terminated or is in a state that rejects epoch alignment (e.g. already started or stopped).","triggerScenarios":"Calling `ack_aligned_initial_epoch` on a handle whose writer was dropped/closed (send to writer fails), or whose state machine is not in the pre-start phase that accepts an aligned initial epoch.","commonSituations":"Writer task crashed earlier during sink recovery; acknowledgment sent after handles were already stopped by a parallelism change; race between failover and sink re-initialization.","solutions":["Capture the inner error (replace `map_err(|_| ...)`) to identify whether the writer channel is closed or the state is wrong.","Confirm the writer task for each handle is running before aligning epochs.","Re-run the initialization flow (`wait_init_handles` then align) if handles were recreated.","Guard against concurrent stop/alter_parallelisms while alignment is in progress."],"exampleFix":"// before\nhandle.ack_aligned_initial_epoch(epoch).map_err(|_| anyhow!(\"failed to ack aligned initial epoch ...\"))?;\n\n// after\nhandle.ack_aligned_initial_epoch(epoch)\n    .with_context(|| format!(\"failed to ack aligned initial epoch {} for handle {:?}\", epoch, handle_id))?;","handlingStrategy":"try-catch","validationCode":"// Ensure all handles are alive before acking alignment\nfor id in manager.registered_handle_ids() {\n    anyhow::ensure!(manager.is_handle_alive(&id), \"handle {} not alive\", id);\n}","typeGuard":null,"tryCatchPattern":"match manager.ack_aligned_initial_epoch(epoch) {\n    Ok(()) => {},\n    Err(e) if e.to_string().contains(\"failed to ack aligned initial epoch\") => {\n        warn!(error = ?e, \"re-running init flow for sink alignment\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Verify writer tasks are running before epoch alignment","Suppress stop/parallelism-change while alignment is in flight","Capture the inner handle error for diagnosis"],"tags":["rust","recovery","writer-actor","sink-coordination"],"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"}