{"record":{"id":"f7a8bf659f623d0c","repo":"risingwavelabs/risingwave","slug":"failed-to-find-handle-to-start","errorCode":null,"errorMessage":"failed to find handle {} to start","messagePattern":"failed to find handle (.+?) to start","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":293,"sourceCode":"\nstruct CoordinationHandleManager {\n    param: SinkParam,\n    writer_handles: HashMap<HandleId, SinkWriterCoordinationHandle>,\n    next_handle_id: HandleId,\n    request_rx: UnboundedReceiver<SinkWriterCoordinationHandle>,\n}\n\nimpl CoordinationHandleManager {\n    fn start(\n        &mut self,\n        log_store_rewind_start_epoch: Option<u64>,\n        handle_ids: impl IntoIterator<Item = HandleId>,\n    ) -> anyhow::Result<()> {\n        for handle_id in handle_ids {\n            let handle = self\n                .writer_handles\n                .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 {}\",","sourceCodeStart":275,"sourceCodeEnd":311,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L275-L311","documentation":"`CoordinationHandleManager::start` is asked to start a set of writer handles by `HandleId`, but the id is not present in `self.writer_handles`. Since every valid handle is registered in `start_handle` (via `next_event`) before it can be started, a missing id means the caller passed stale or fabricated handle ids — an internal bookkeeping inconsistency between the caller (e.g. `wait_init_handles`) and the manager's registry.","triggerScenarios":"Calling `start(handle_ids, log_store_rewind_start_epoch)` with a `HandleId` that was never inserted into `writer_handles`, or one that was already consumed/removed (e.g. after `Stop`), or a resurrected job whose handles were dropped on a previous error path.","commonSituations":"Sink job recovery after meta-node restart where persisted handle ids no longer match live registrations; race between parallelism change (dropping handles) and a concurrent start; bugs in handle-id persistence/migration code.","solutions":["Log the full set of keys in `writer_handles` versus the requested `handle_ids` to identify the stale id.","Ensure `wait_init_handles` collects handle ids only from `next_event` results (live registrations).","If the sink job is being recovered, re-register the writer handles before calling `start`.","Check whether a concurrent `alter_parallelisms`/stop path removed the handle while start was in flight and add synchronization."],"exampleFix":"// before\nlet handle_ids: Vec<HandleId> = persisted_ids.clone();\nmanager.start(handle_ids, epoch)?;\n\n// after\nlet handle_ids: Vec<HandleId> = manager\n    .registered_handle_ids() // ids taken from live writer_handles\n    .collect();\nmanager.start(handle_ids, epoch)?;","handlingStrategy":"validation","validationCode":"// Validate handle ids before calling start\nlet registered: HashSet<_> = manager.registered_handle_ids().collect();\nlet missing: Vec<_> = handle_ids.iter().filter(|id| !registered.contains(id)).collect();\nanyhow::ensure!(missing.is_empty(), \"unregistered handle ids: {:?}\", missing);","typeGuard":"fn is_registered(mgr: &CoordinationHandleManager, id: &HandleId) -> bool {\n    mgr.registered_handle_ids().any(|rid| rid == *id)\n}","tryCatchPattern":"match manager.start(handle_ids, epoch) {\n    Ok(()) => {},\n    Err(e) if e.to_string().contains(\"failed to find handle\") => {\n        warn!(\"stale handle ids; re-collecting from live registrations\");\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Only derive HandleIds from next_event results, never from persisted metadata","Log writer_handles keys when start fails","Serialize parallelism changes with start operations"],"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"}