risingwavelabs/risingwave · error · SinkError

channel closed

Error message

channel closed

What it means

Data is streamed to the BE through an unbounded mpsc channel whose receiver is consumed by the HTTP request body in a spawned task. If the channel's receiving end is dropped — the HTTP request has already failed or completed — `send` on the channel returns Err and this error is thrown after awaiting the join handle to surface the underlying request error.

Source

Thrown at src/connector/src/sink/doris_starrocks_connector.rs:361

            join_handle,
            buffer: BytesMut::with_capacity(BUFFER_SIZE),
            stream_load_http_timeout,
        }
    }

    async fn send_chunk(&mut self) -> Result<()> {
        if self.sender.is_none() {
            return Ok(());
        }

        let chunk = mem::replace(&mut self.buffer, BytesMut::with_capacity(BUFFER_SIZE));

        match self.sender.as_mut().unwrap().send(chunk.freeze()) {
            Err(_e) => {
                self.sender.take();
                self.wait_handle().await?;

                Err(SinkError::DorisStarrocksConnect(anyhow!("channel closed")))
            }
            _ => Ok(()),
        }
    }

    pub async fn write(&mut self, data: Bytes) -> Result<()> {
        self.buffer.put_slice(&data);
        if self.buffer.len() >= MIN_CHUNK_SIZE {
            self.send_chunk().await?;
        }
        Ok(())
    }

    async fn wait_handle(&mut self) -> Result<Vec<u8>> {
        let res = match tokio::time::timeout(self.stream_load_http_timeout, &mut self.join_handle)
            .await
        {
            Ok(res) => res.map_err(|err| SinkError::DorisStarrocksConnect(anyhow!(err)))??,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the error surfaced by `wait_handle` in the same error chain — it holds the real HTTP failure from the BE request
2. Verify network stability and BE health between RisingWave and the Doris/StarRocks BE
3. Increase `stream_load_http_timeout` if large chunks exceed the configured timeout
4. Reduce chunk/buffer pressure or retry the sink write — the inserter marks the sender as taken after this error, so the sink commit will fail and can be retried
5. Inspect BE logs for stream-load aborts around the failure time

match res {
    Err(e) if e.to_string().contains("channel closed") => {
        // the BE request failed earlier — inspect chained error, then retry the chunk/commit
    }
    r => r?,
}

When it happens

Trigger: `send_chunk()` (reached from `write` when the buffer exceeds MIN_CHUNK_SIZE, or from `finish`) sends buffered bytes while the spawned stream-load request has already terminated (connection error, non-OK status, panic), dropping the receiver.

Common situations: BE closed the connection mid-load (timeout, crash, network interruption); the initial request failed before streaming began; a previous chunk's failure left the sender orphaned; stream_load_http_timeout elapsed and the task was aborted.

Related errors

• Failed connection {:?},{:?}
• failed to parse response body
• sending stream load request failed
• Doris/Starrocks connect error: {0}
• Doris error: {0}

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