{"record":{"id":"c223be3ef11ff9d2","repo":"risingwavelabs/risingwave","slug":"failed-to-find-handle-when-acknowledging-the-co","errorCode":null,"errorMessage":"failed to find handle {} when acknowledging the commit for epoch {}","messagePattern":"failed to find handle (.+?) when acknowledging the commit for epoch (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":327,"sourceCode":"                .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 {}\",\n                    handle_id,\n                    epoch\n                )\n            })?;\n            handle.ack_commit(epoch).map_err(|_| {\n                anyhow!(\n                    \"failed to acknowledge the commit for epoch {} on handle {}\",\n                    epoch,\n                    handle_id\n                )\n            })?;\n        }\n        Ok(())\n    }\n\n    async fn next_request_inner(\n        writer_handles: &mut HashMap<HandleId, SinkWriterCoordinationHandle>,","sourceCodeStart":309,"sourceCodeEnd":345,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L309-L345","documentation":"`ack_commit` looks up each supplied `HandleId` in `writer_handles` to acknowledge a committed epoch; this error means one of the ids was not found. As with error 1772, this signals stale or unregistered handle ids reaching the manager — the commit was for a sink writer the coordinator no longer tracks.","triggerScenarios":"A commit request was accepted from a handle that was subsequently removed (e.g. after `Stop` or during `alter_parallelisms`), and its id is later passed to `ack_commit`; or the caller passes ids from persisted metadata that do not match live registrations.","commonSituations":"Parallelism change concurrently dropping writers while their epochs are being committed; recovery from a snapshot where handle ids were reassigned; double commit acknowledgments after failover.","solutions":["Filter the ids passed to `ack_commit` against currently registered handles before calling it.","Track per-handle pending epochs so commits for removed handles are ignored rather than erroring.","Serialize stop/parallelism-change operations against commit acknowledgment to avoid the race.","If the sink is recovering, re-register handles before acknowledging commits."],"exampleFix":"// before\nmanager.ack_commit(epoch, all_pending_ids)?;\n\n// after\nlet live: HashSet<HandleId> = manager.registered_handle_ids().collect();\nmanager.ack_commit(epoch, all_pending_ids.into_iter().filter(|id| live.contains(id)))?;","handlingStrategy":"validation","validationCode":"// Drop ids for handles that no longer exist before acking\nlet live: HashSet<_> = manager.registered_handle_ids().collect();\nlet ackable: Vec<_> = pending_ids.into_iter().filter(|id| live.contains(id)).collect();","typeGuard":"fn handle_exists(mgr: &CoordinationHandleManager, id: &HandleId) -> bool {\n    mgr.registered_handle_ids().contains(id)\n}","tryCatchPattern":"if let Err(e) = manager.ack_commit(epoch, ids) {\n    if e.to_string().contains(\"failed to find handle\") {\n        warn!(epoch, \"commit for removed handle ignored\");\n    } else { return Err(e); }\n}","preventionTips":["Track pending epochs per handle and drop entries when handles are removed","Serialize stop operations against commit acks","Ignore (don't error) commits for unregistered handles"],"tags":["rust","state-management","handle-not-found","sink-coordination"],"backgroundTag":"record-not-found","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"}