risingwavelabs/risingwave · warning · anyhow::Error

failed to send the commit response for epoch {}

Error message

failed to send the commit response for epoch {}

What it means

This error is raised in the sink coordinator's `ack_commit` when it tries to send a successful `CommitResponse` back to the sink executor over the response channel and the receiver end has already been dropped. It means the gRPC stream or worker connected to this coordinator handle is gone, so the commit acknowledgement can no longer be delivered. The underlying `anyhow!` error discards the `SendError` and only carries the epoch for context.

Source

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

                msg: Some(coordinate_response::Msg::AlignInitialEpochResponse(
                    aligned_initial_epoch,
                )),
            }))
            .map_err(|_| anyhow!("failed to send the start response"))
    }

    pub(super) fn abort(self, status: Status) {
        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)?;

View on GitHub (pinned to 6469eb736d)

Solutions

1. Treat this as a disconnected client: log at debug/warn level and let the coordinator worker terminate; the sink executor will re-connect and re-request.
2. Check sink coordinator logs immediately before this error for stream closure or worker panic to find why `response_tx` lost its receiver.
3. Ensure the sink coordinator manager keeps the receiver alive until the worker future completes (poll the worker until `Poll::Ready(())` before dropping it).
4. If it recurs during failover, upgrade to a version with graceful stream shutdown and verify client-side retry logic re-establishes coordination.

Example fix

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

// Check the channel still has its receiver before acking
if response_tx.is_closed() {
    tracing::warn!(epoch, "commit response receiver gone; skipping ack");
    return;
}

// Treat send failure as client disconnect, not a fatal error
if let Err(e) = handle.ack_commit(epoch) {
    tracing::debug!("commit response not delivered (client gone): {e:#}");
}

When it happens

Trigger: The sink coordinator manager drops the response receiver (worker shut down, gRPC stream closed, panic in the worker) before or while `ack_commit` is called to reply to a CommitRequest for the given epoch.

Common situations: User cancels a `DROP SINK` or the sink's streaming job fails over while the coordinator is mid-handshake; network disconnect between frontend/sink worker and meta node closes the coordinate stream; meta node restarts during a commit epoch transition.

Related errors

• failed to send the stopped response
• 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/37b5e2f975f0a635. Report an issue: GitHub.