risingwavelabs/risingwave · info · anyhow::Error

end of request stream

Error message

end of request stream

What it means

`poll_next_request` returns this error when the incoming `CoordinateRequest` stream yields `None`, meaning the peer closed the request stream. The coordinator cannot serve any more requests, so it surfaces the end-of-stream as an error to terminate the coordination worker. Normally a deliberate client disconnect, not a data corruption issue.

Source

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

            }))
            .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
                    {
                        return Poll::Ready(Err(anyhow!(
                            "invalid commit epoch {}, prev_epoch {}",
                            request.epoch,
                            prev_epoch
                        )));
                    }
                    if request.metadata.is_none() {
                        return Poll::Ready(Err(anyhow!("empty commit metadata")));

View on GitHub (pinned to 6469eb736d)

Solutions

1. Expected during failover/shutdown: treat as a clean disconnect, log at info/debug, and exit the coordinator worker loop without alerting.
2. Check for sink executor crashes or OOM kills around the timestamp of this error if it happens unexpectedly.
3. Enable keepalive on the gRPC channel to prevent LBs from closing idle coordinate streams.
4. Verify client-side code keeps the stream open for the lifetime of the sink's coordination session.

// Detect stream health before relying on it
if request_stream_healthy == false {
    tracing::info!("coordinate stream closed; awaiting reconnect");
    return Ok(());
}

match handle.poll_next_request(cx) {
    Poll::Ready(Err(e)) if e.to_string().contains("end of request stream") => {
        tracing::info!("client closed coordinate stream; exiting worker");
    }
    other => { /* normal handling */ }
}

When it happens

Trigger: The gRPC `coordinate` stream from the sink executor ends (client dropped the stream, network closed, executor terminated) while the coordinator worker is still polling for the next request.

Common situations: Streaming job failover or cancellation closes the coordinate channel; frontend/node restart; idle connection reaped by a load balancer between sink requests.

Related errors

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

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