{"record":{"id":"1003b022d541136c","repo":"risingwavelabs/risingwave","slug":"failed-to-start-for-handle","errorCode":null,"errorMessage":"failed to start {:?} for handle {}","messagePattern":"failed to start (.+?) for handle (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":295,"sourceCode":"    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 {}\",\n                        aligned_initial_epoch,\n                        handle_id","sourceCodeStart":277,"sourceCodeEnd":313,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L277-L313","documentation":"The handle exists in `writer_handles`, but its `start(log_store_rewind_start_epoch)` call returned `Err`. The manager discards the handle's own error and wraps it in this generic message, so the failure happened inside the coordination handle's start routine — typically rewinding/joining the writer to the given log-store epoch or spawning its task.","triggerScenarios":"Calling `start` on a registered handle whose underlying writer channel/task is already closed, or where the log store cannot rewind to `log_store_rewind_start_epoch` (epoch no longer retained), or the handle was already started/stopped.","commonSituations":"Recovery after a barrier/log-store truncation removed the requested start epoch; writer actor already exited due to an earlier error; double start of the same sink job during failover.","solutions":["Preserve and log the inner `Err` from `handle.start` (change `map_err(|_| ...)` to include the source) to see the root cause.","Verify `log_store_rewind_start_epoch` is still available in the log store and not truncated.","Ensure the writer handle's task/channel is alive before starting; recreate the handle if it was dropped.","Avoid double-starting the same handle across recovery attempts; check for idempotency in `handle.start`."],"exampleFix":"// before\nhandle.start(log_store_rewind_start_epoch).map_err(|_| {\n    anyhow!(\"failed to start {:?} for handle {}\", log_store_rewind_start_epoch, handle_id)\n})?;\n\n// after\nhandle.start(log_store_rewind_start_epoch)\n    .with_context(|| format!(\"failed to start {:?} for handle {}\", log_store_rewind_start_epoch, handle_id))?;","handlingStrategy":"try-catch","validationCode":"// Check the epoch is still retained in the log store before starting\nanyhow::ensure!(\n    log_store.contains_epoch(log_store_rewind_start_epoch),\n    \"rewind epoch {} no longer retained\",\n    log_store_rewind_start_epoch\n);","typeGuard":null,"tryCatchPattern":"if let Err(e) = manager.start(ids, rewind_epoch) {\n    error!(error = ?e, \"handle start failed; will retry after re-registration\");\n    // re-register handles and retry once before failing the job\n}","preventionTips":["Never discard inner errors with map_err(|_| ...); use with_context","Confirm the rewind epoch is within log-store retention","Avoid double-starting the same sink job during failover"],"tags":["rust","recovery","log-store","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"}