What causes "end of writer request stream" in risingwavelabs/risingwave?

This is usually an expected shutdown signal: treat it as job termination and clean up coordination state gracefully.

end of writer request stream

Error message

end of writer request stream

What it means

`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.

Source

Thrown at src/meta/src/manager/sink_coordination/coordinator_worker.rs:387

}

impl CoordinationHandleManagerEvent {
    fn name(&self) -> &'static str {
        match self {
            CoordinationHandleManagerEvent::NewHandle => "NewHandle",
            CoordinationHandleManagerEvent::UpdateVnodeBitmap => "UpdateVnodeBitmap",
            CoordinationHandleManagerEvent::Stop => "Stop",
            CoordinationHandleManagerEvent::CommitRequest { .. } => "CommitRequest",
            CoordinationHandleManagerEvent::AlignInitialEpoch(_) => "AlignInitialEpoch",
        }
    }
}

impl CoordinationHandleManager {
    async fn next_event(&mut self) -> anyhow::Result<(HandleId, CoordinationHandleManagerEvent)> {
        select! {
            handle = self.request_rx.recv() => {
                let handle = handle.ok_or_else(|| anyhow!("end of writer request stream"))?;
                if handle.param() != &self.param {
                    warn!(prev_param = ?self.param, new_param = ?handle.param(), "sink param mismatch");
                }
                let handle_id = self.next_handle_id;
                self.next_handle_id += 1;
                self.writer_handles.insert(handle_id, handle);
                Ok((handle_id, CoordinationHandleManagerEvent::NewHandle))
            }
            result = Self::next_request_inner(&mut self.writer_handles) => {
                let (handle_id, request) = result?;
                let event = match request {
                    coordinate_request::Msg::CommitRequest(request) => {
                        CoordinationHandleManagerEvent::CommitRequest {
                            epoch: request.epoch,
                            metadata: request.metadata.ok_or_else(|| anyhow!("empty sink metadata"))?,
                            schema_change: request.schema_change,
                        }
                    }

View on GitHub (pinned to 6469eb736d)

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.

Example fix

// before
let (handle_id, event) = manager.next_event().await?;

// after
match manager.next_event().await {
    Ok((handle_id, event)) => { /* normal coordination */ }
    Err(e) if e.to_string().contains("end of writer request stream") => {
        info!(sink_id = %sink_id, "all writers gone; stopping coordination");
        return Ok(()); // graceful termination
    }
    Err(e) => return Err(e),
}

Defensive patterns

Strategy: try-catch

Validate before calling

// Confirm at least one writer is connected before entering the coordination loop
anyhow::ensure!(
    !manager.registered_handle_ids().is_empty() || has_active_senders(&request_rx),
    "no sink writers connected to coordination channel"
);

Try / catch

match manager.next_event().await {
    Err(e) if is_end_of_writer_stream(&e) => {
        info!("all sink writers gone; terminating coordination gracefully");
        return Ok(());
    }
    result => result?,
}

Prevention

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

When it happens

Trigger: 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.

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

Related errors

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