{"record":{"id":"355f65d9f9485ef1","repo":"risingwavelabs/risingwave","slug":"end-of-request-stream","errorCode":null,"errorMessage":"end of request stream","messagePattern":"end of request stream","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"info","filePath":"src/meta/src/manager/sink_coordination/handle.rs","lineNumber":119,"sourceCode":"            }))\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\n                    {\n                        return Poll::Ready(Err(anyhow!(\n                            \"invalid commit epoch {}, prev_epoch {}\",\n                            request.epoch,\n                            prev_epoch\n                        )));\n                    }\n                    if request.metadata.is_none() {\n                        return Poll::Ready(Err(anyhow!(\"empty commit metadata\")));","sourceCodeStart":101,"sourceCodeEnd":137,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/handle.rs#L101-L137","documentation":"`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.","triggerScenarios":"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.","commonSituations":"Streaming job failover or cancellation closes the coordinate channel; frontend/node restart; idle connection reaped by a load balancer between sink requests.","solutions":["Expected during failover/shutdown: treat as a clean disconnect, log at info/debug, and exit the coordinator worker loop without alerting.","Check for sink executor crashes or OOM kills around the timestamp of this error if it happens unexpectedly.","Enable keepalive on the gRPC channel to prevent LBs from closing idle coordinate streams.","Verify client-side code keeps the stream open for the lifetime of the sink's coordination session."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Detect stream health before relying on it\nif request_stream_healthy == false {\n    tracing::info!(\"coordinate stream closed; awaiting reconnect\");\n    return Ok(());\n}","typeGuard":null,"tryCatchPattern":"match handle.poll_next_request(cx) {\n    Poll::Ready(Err(e)) if e.to_string().contains(\"end of request stream\") => {\n        tracing::info!(\"client closed coordinate stream; exiting worker\");\n    }\n    other => { /* normal handling */ }\n}","preventionTips":["Configure gRPC keepalive to survive idle periods behind load balancers","Design the coordinator to accept clean stream termination without alerting","Monitor executor crashes/OOM separately from normal disconnects"],"tags":["rust","meta","grpc-stream","end-of-stream","sink-coordinator"],"backgroundTag":"unexpected-response-shape","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"}