risingwavelabs/risingwave · warning · anyhow::Error

failed to send the stopped response

Error message

failed to send the stopped response

What it means

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.

Source

Thrown at src/meta/src/manager/sink_coordination/handle.rs:110

        let _ = self.response_tx.send(Err(status));
    }

    pub(super) fn ack_commit(&mut self, epoch: u64) -> anyhow::Result<()> {
        self.response_tx
            .send(Ok(CoordinateResponse {
                msg: Some(coordinate_response::Msg::CommitResponse(CommitResponse {
                    epoch,
                })),
            }))
            .map_err(|_| anyhow!("failed to send the commit response for epoch {}", epoch))
    }

    pub(super) fn stop(&mut self) -> anyhow::Result<()> {
        self.response_tx
            .send(Ok(CoordinateResponse {
                msg: Some(coordinate_response::Msg::Stopped(true)),
            }))
            .map_err(|_| anyhow!("failed to send the stopped response"))
    }

    pub(super) fn poll_next_request(
        &mut self,
        cx: &mut Context<'_>,
    ) -> Poll<anyhow::Result<coordinate_request::Msg>> {
        let result = try {
            let request = ready!(self.request_stream.try_poll_next_unpin(cx))
                .ok_or_else(|| anyhow!("end of request stream"))?
                .map_err(anyhow::Error::from)?;
            let request = request.msg.ok_or_else(|| anyhow!("None msg in request"))?;
            match &request {
                coordinate_request::Msg::StartRequest(_)
                | coordinate_request::Msg::Stop(_)
                | coordinate_request::Msg::AlignInitialEpochRequest(_) => {}
                coordinate_request::Msg::CommitRequest(request) => {
                    if let Some(prev_epoch) = self.prev_epoch
                        && request.epoch < prev_epoch

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the caller keeps the response receiver alive until it has read the `Stopped` reply.
2. Log and ignore: a failure to notify a dead client about a stop is usually benign; the coordinator is being shut down anyway.
3. Check for premature worker exit (panics, early return in the coordinator loop) that drops the receiver before stop completes.
4. If stopping via `DROP SINK`, confirm the sink coordinator manager's shutdown ordering stops the worker only after the stop response is consumed.

Example fix

// before
.map_err(|_| anyhow!("failed to send the stopped response"))
// after
.map_err(|e| anyhow!("failed to send the stopped response: {e}"))

if response_tx.is_closed() {
    tracing::debug!("receiver already gone; stop response not needed");
    return Ok(());
}

match handle.stop() {
    Ok(()) => tracing::debug!("coordinator stopped cleanly"),
    Err(e) => tracing::debug!("stop response undeliverable (client gone): {e:#}"),
}

When it happens

Trigger: Calling `SinkCoordinatorHandle::stop` after the worker/gRPC handler that owns the response receiver has already exited (stream cancelled, client disconnected, worker task finished).

Common situations: 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.

Related errors

• failed to send the commit response for epoch {}
• end of request stream
• None msg in request
• empty vnode bitmap
• reschedule failed

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/bced67f1c02022c4. Report an issue: GitHub.