{"record":{"id":"bced67f1c02022c4","repo":"risingwavelabs/risingwave","slug":"failed-to-send-the-stopped-response","errorCode":null,"errorMessage":"failed to send the stopped response","messagePattern":"failed to send the stopped response","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"warning","filePath":"src/meta/src/manager/sink_coordination/handle.rs","lineNumber":110,"sourceCode":"        let _ = self.response_tx.send(Err(status));\n    }\n\n    pub(super) fn ack_commit(&mut self, epoch: u64) -> anyhow::Result<()> {\n        self.response_tx\n            .send(Ok(CoordinateResponse {\n                msg: Some(coordinate_response::Msg::CommitResponse(CommitResponse {\n                    epoch,\n                })),\n            }))\n            .map_err(|_| anyhow!(\"failed to send the commit response for epoch {}\", epoch))\n    }\n\n    pub(super) fn stop(&mut self) -> anyhow::Result<()> {\n        self.response_tx\n            .send(Ok(CoordinateResponse {\n                msg: Some(coordinate_response::Msg::Stopped(true)),\n            }))\n            .map_err(|_| anyhow!(\"failed to send the stopped response\"))\n    }\n\n    pub(super) fn poll_next_request(\n        &mut self,\n        cx: &mut Context<'_>,\n    ) -> Poll<anyhow::Result<coordinate_request::Msg>> {\n        let result = try {\n            let request = ready!(self.request_stream.try_poll_next_unpin(cx))\n                .ok_or_else(|| anyhow!(\"end of request stream\"))?\n                .map_err(anyhow::Error::from)?;\n            let request = request.msg.ok_or_else(|| anyhow!(\"None msg in request\"))?;\n            match &request {\n                coordinate_request::Msg::StartRequest(_)\n                | coordinate_request::Msg::Stop(_)\n                | coordinate_request::Msg::AlignInitialEpochRequest(_) => {}\n                coordinate_request::Msg::CommitRequest(request) => {\n                    if let Some(prev_epoch) = self.prev_epoch\n                        && request.epoch < prev_epoch","sourceCodeStart":92,"sourceCodeEnd":128,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/handle.rs#L92-L128","documentation":"Raised in `stop` when the coordinator attempts to send a `Stopped(true)` response to acknowledge shutdown, but the receiving side of `response_tx` has been dropped. The client that asked to stop the coordination will never receive this acknowledgement. Like the other channel errors here, the real `SendError` cause is swallowed.","triggerScenarios":"Calling `SinkCoordinatorHandle::stop` after the worker/gRPC handler that owns the response receiver has already exited (stream cancelled, client disconnected, worker task finished).","commonSituations":"Client cancels the coordination RPC while the coordinator is processing the stop; sink coordinator worker completes earlier than expected; double-stop sequences where the first stop already tore down the channel.","solutions":["Verify the caller keeps the response receiver alive until it has read the `Stopped` reply.","Log and ignore: a failure to notify a dead client about a stop is usually benign; the coordinator is being shut down anyway.","Check for premature worker exit (panics, early return in the coordinator loop) that drops the receiver before stop completes.","If stopping via `DROP SINK`, confirm the sink coordinator manager's shutdown ordering stops the worker only after the stop response is consumed."],"exampleFix":"// before\n.map_err(|_| anyhow!(\"failed to send the stopped response\"))\n// after\n.map_err(|e| anyhow!(\"failed to send the stopped response: {e}\"))","handlingStrategy":"try-catch","validationCode":"if response_tx.is_closed() {\n    tracing::debug!(\"receiver already gone; stop response not needed\");\n    return Ok(());\n}","typeGuard":null,"tryCatchPattern":"match handle.stop() {\n    Ok(()) => tracing::debug!(\"coordinator stopped cleanly\"),\n    Err(e) => tracing::debug!(\"stop response undeliverable (client gone): {e:#}\"),\n}","preventionTips":["Send the Stopped ack before tearing down the worker's receiver","Make stop idempotent and treat undeliverable acks as success","Avoid double-stop sequences; track stop state explicitly"],"tags":["rust","meta","sink-coordinator","channel-closed","shutdown"],"backgroundTag":"broken-pipe","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"}