risingwavelabs/risingwave · error · SinkError::Remote

unable to send response

Error message

unable to send response: {:?}

What it means

In the log-sink writer loop, the remote sink tries to forward each response received from the remote writer stream to response_tx (the channel back to the stream executor). If the receiving end of that channel has already been dropped (receiver gone), send fails and this error is raised.

Solutions

1. Typically benign during shutdown/recovery; check whether the sink actor was restarted and whether the error is recurring.
2. If recurring, investigate why the response receiver exits before the writer finishes (look at executor logs just before this error).
3. Retry the sink after recovery completes; RisingWave will rebuild the channel.

if let Err(e) = result {
    if e.to_string().contains("unable to send response") {
        // receiver dropped; treat as shutdown path, don't crash the actor
        return Ok(()); // or propagate depending on recovery semantics
    }
    return Err(e);
}

When it happens

Trigger: consume_log_and_sink receives Ok(Some(response)) from the log reader, but response_tx.send fails because the downstream receiver task was dropped/aborted (e.g. executor shutdown, stream aborted before draining responses).

Common situations: Actor/executor is being shut down or migrated during recovery while the sink writer task is still pushing responses; a panic or early return in the consumer side; barrier aborts that drop the channel early.

Related errors

• Hummock committed epoch sender closed unexpectedly
• Lance fragment write task stopped before accepting a record…
• channel closed
• (dynamic: channel send error on shutdown signal)
• failed to receive rebuild sink notify

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

Appendix: source

Thrown at src/connector/src/sink/remote.rs:346

    async fn consume_log_and_sink(self, mut log_reader: impl SinkLogReader) -> Result<!> {
        log_reader.start_from(None).await?;
        let mut request_tx = self.request_sender;
        let mut response_err_stream_rx = self.response_stream;
        let sink_writer_metrics = self.sink_writer_metrics;

        let (response_tx, mut response_rx) = unbounded_channel();

        let poll_response_stream = async move {
            loop {
                let result = response_err_stream_rx
                    .stream
                    .try_next()
                    .instrument_await("log_sinker_wait_next_response")
                    .await;
                match result {
                    Ok(Some(response)) => {
                        response_tx.send(response).map_err(|err| {
                            SinkError::Remote(anyhow!("unable to send response: {:?}", err.0))
                        })?;
                    }
                    Ok(None) => return Err(SinkError::Remote(anyhow!("end of response stream"))),
                    Err(e) => return Err(SinkError::Remote(anyhow!(e))),
                }
            }
        };

        let poll_consume_log_and_sink = async move {
            fn truncate_matched_offset(
                queue: &mut VecDeque<(TruncateOffset, Option<Instant>)>,
                persisted_offset: TruncateOffset,
                log_reader: &mut impl SinkLogReader,
                sink_writer_metrics: &SinkWriterMetrics,
            ) -> Result<()> {
                while let Some((sent_offset, _)) = queue.front()
                    && sent_offset < &persisted_offset
                {

View on GitHub (pinned to 6469eb736d)