{"record":{"id":"162c428bddfd3b1e","repo":"risingwavelabs/risingwave","slug":"failed-to-acknowledge-the-commit-for-epoch-on-h","errorCode":null,"errorMessage":"failed to acknowledge the commit for epoch {} on handle {}","messagePattern":"failed to acknowledge the commit for epoch (.+?) on handle (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":334,"sourceCode":"        }\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>,\n    ) -> anyhow::Result<(HandleId, coordinate_request::Msg)> {\n        poll_fn(|cx| {\n            for (handle_id, handle) in writer_handles.iter_mut() {\n                if let Poll::Ready(result) = handle.poll_next_request(cx) {\n                    return Poll::Ready(result.map(|request| (*handle_id, request)));\n                }\n            }","sourceCodeStart":316,"sourceCodeEnd":352,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L316-L352","documentation":"The handle exists, but its `ack_commit(epoch)` returned `Err`; the manager wraps this into the given message, discarding the inner cause. Typical root causes are the writer actor being gone (channel closed) or the epoch not matching the writer's expectations (e.g. acking an epoch that was not prepared or was already acked).","triggerScenarios":"Calling `ack_commit` on a live-registered handle whose writer task exited, or acking an epoch out of order (not prepared / already committed), during commit processing in the coordination worker.","commonSituations":"Writer crashed mid-transaction so the commit ack cannot be delivered; duplicated commit events after a retry/failover; epochs committed after the handle was logically stopped but before removal.","solutions":["Include the inner error in the wrapper (use `with_context` instead of `map_err(|_| ...)`) to expose the real cause.","Verify the epoch being acked was prepared on that handle and not previously acked.","Check writer task health/logs for a crash preceding the failed ack.","Make `ack_commit` on the handle idempotent or tolerant of repeated epochs to survive event replays."],"exampleFix":"// before\nhandle.ack_commit(epoch).map_err(|_| anyhow!(\"failed to acknowledge the commit for epoch {} on handle {}\", epoch, handle_id))?;\n\n// after\nhandle.ack_commit(epoch)\n    .with_context(|| format!(\"failed to acknowledge the commit for epoch {} on handle {}\", epoch, handle_id))?;","handlingStrategy":"try-catch","validationCode":"// Only ack epochs that were prepared on this handle and not yet acked\nanyhow::ensure!(\n    prepared_epochs.contains(&epoch) && !acked_epochs.contains(&epoch),\n    \"epoch {} not prepared or already acked\",\n    epoch\n);","typeGuard":null,"tryCatchPattern":"match handle.ack_commit(epoch) {\n    Ok(()) => {},\n    Err(e) => {\n        warn!(epoch, error = ?e, \"commit ack failed; writer may have exited\");\n        // recreate handle or fail the sink job with the preserved cause\n    }\n}","preventionTips":["Make ack_commit idempotent for repeated epochs","Monitor writer actor health before commit acks","Preserve inner errors instead of map_err(|_| ...)"],"tags":["rust","commit-ack","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"}