{"record":{"id":"21defc502e0b0e1a","repo":"risingwavelabs/risingwave","slug":"end-of-writer-request-stream","errorCode":null,"errorMessage":"end of writer request stream","messagePattern":"end of writer request stream","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/sink_coordination/coordinator_worker.rs","lineNumber":387,"sourceCode":"}\n\nimpl CoordinationHandleManagerEvent {\n    fn name(&self) -> &'static str {\n        match self {\n            CoordinationHandleManagerEvent::NewHandle => \"NewHandle\",\n            CoordinationHandleManagerEvent::UpdateVnodeBitmap => \"UpdateVnodeBitmap\",\n            CoordinationHandleManagerEvent::Stop => \"Stop\",\n            CoordinationHandleManagerEvent::CommitRequest { .. } => \"CommitRequest\",\n            CoordinationHandleManagerEvent::AlignInitialEpoch(_) => \"AlignInitialEpoch\",\n        }\n    }\n}\n\nimpl CoordinationHandleManager {\n    async fn next_event(&mut self) -> anyhow::Result<(HandleId, CoordinationHandleManagerEvent)> {\n        select! {\n            handle = self.request_rx.recv() => {\n                let handle = handle.ok_or_else(|| anyhow!(\"end of writer request stream\"))?;\n                if handle.param() != &self.param {\n                    warn!(prev_param = ?self.param, new_param = ?handle.param(), \"sink param mismatch\");\n                }\n                let handle_id = self.next_handle_id;\n                self.next_handle_id += 1;\n                self.writer_handles.insert(handle_id, handle);\n                Ok((handle_id, CoordinationHandleManagerEvent::NewHandle))\n            }\n            result = Self::next_request_inner(&mut self.writer_handles) => {\n                let (handle_id, request) = result?;\n                let event = match request {\n                    coordinate_request::Msg::CommitRequest(request) => {\n                        CoordinationHandleManagerEvent::CommitRequest {\n                            epoch: request.epoch,\n                            metadata: request.metadata.ok_or_else(|| anyhow!(\"empty sink metadata\"))?,\n                            schema_change: request.schema_change,\n                        }\n                    }","sourceCodeStart":369,"sourceCodeEnd":405,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/sink_coordination/coordinator_worker.rs#L369-L405","documentation":"`CoordinationHandleManager::next_event` multiplexes between receiving new writer handles and requests from existing ones. When `request_rx.recv()` returns `None`, all senders of the writer request stream are gone, and this error is raised. It means no sink writer remains connected to the coordinator, so the coordination loop cannot proceed and is expected to terminate the sink job.","triggerScenarios":"All sink writer actors dropped their request senders — e.g. every writer failed, the sink job was cancelled/dropped, or actors exited during failover — while the coordinator's `next_event` (called from `wait_init_handles` or `alter_parallelisms`) was still waiting.","commonSituations":"Dropping a materialized sink while coordination is in progress; all writer tasks crashing (OOM, panic, actor failure); shutting down compute nodes hosting the sink writers.","solutions":["This is usually an expected shutdown signal: treat it as job termination and clean up coordination state gracefully.","If unexpected, check compute-node logs for writer actor crashes (panics, OOM) preceding this error.","Verify the sink job's fragment/actors are running and connected to the meta node's coordination channel.","Recover the sink job (recreate/restart) so writers reconnect and re-register their handles."],"exampleFix":"// before\nlet (handle_id, event) = manager.next_event().await?;\n\n// after\nmatch manager.next_event().await {\n    Ok((handle_id, event)) => { /* normal coordination */ }\n    Err(e) if e.to_string().contains(\"end of writer request stream\") => {\n        info!(sink_id = %sink_id, \"all writers gone; stopping coordination\");\n        return Ok(()); // graceful termination\n    }\n    Err(e) => return Err(e),\n}","handlingStrategy":"try-catch","validationCode":"// Confirm at least one writer is connected before entering the coordination loop\nanyhow::ensure!(\n    !manager.registered_handle_ids().is_empty() || has_active_senders(&request_rx),\n    \"no sink writers connected to coordination channel\"\n);","typeGuard":null,"tryCatchPattern":"match manager.next_event().await {\n    Err(e) if is_end_of_writer_stream(&e) => {\n        info!(\"all sink writers gone; terminating coordination gracefully\");\n        return Ok(());\n    }\n    result => result?,\n}","preventionTips":["Treat empty request stream as normal shutdown, not a hard failure","Monitor writer actor crashes (OOM/panic) on compute nodes","Verify sink actors exist before starting coordination"],"tags":["rust","channel-closed","actor-lifecycle","sink-coordination"],"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"}