{"record":{"id":"b835c57b805b6ab1","repo":"risingwavelabs/risingwave","slug":"channel-closed","errorCode":null,"errorMessage":"channel closed","messagePattern":"channel closed","errorType":"exception","errorClass":"SinkError","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/doris_starrocks_connector.rs","lineNumber":361,"sourceCode":"            join_handle,\n            buffer: BytesMut::with_capacity(BUFFER_SIZE),\n            stream_load_http_timeout,\n        }\n    }\n\n    async fn send_chunk(&mut self) -> Result<()> {\n        if self.sender.is_none() {\n            return Ok(());\n        }\n\n        let chunk = mem::replace(&mut self.buffer, BytesMut::with_capacity(BUFFER_SIZE));\n\n        match self.sender.as_mut().unwrap().send(chunk.freeze()) {\n            Err(_e) => {\n                self.sender.take();\n                self.wait_handle().await?;\n\n                Err(SinkError::DorisStarrocksConnect(anyhow!(\"channel closed\")))\n            }\n            _ => Ok(()),\n        }\n    }\n\n    pub async fn write(&mut self, data: Bytes) -> Result<()> {\n        self.buffer.put_slice(&data);\n        if self.buffer.len() >= MIN_CHUNK_SIZE {\n            self.send_chunk().await?;\n        }\n        Ok(())\n    }\n\n    async fn wait_handle(&mut self) -> Result<Vec<u8>> {\n        let res = match tokio::time::timeout(self.stream_load_http_timeout, &mut self.join_handle)\n            .await\n        {\n            Ok(res) => res.map_err(|err| SinkError::DorisStarrocksConnect(anyhow!(err)))??,","sourceCodeStart":343,"sourceCodeEnd":379,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/doris_starrocks_connector.rs#L343-L379","documentation":"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.","triggerScenarios":"`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.","commonSituations":"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.","solutions":["Check the error surfaced by `wait_handle` in the same error chain — it holds the real HTTP failure from the BE request","Verify network stability and BE health between RisingWave and the Doris/StarRocks BE","Increase `stream_load_http_timeout` if large chunks exceed the configured timeout","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","Inspect BE logs for stream-load aborts around the failure time"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match res {\n    Err(e) if e.to_string().contains(\"channel closed\") => {\n        // the BE request failed earlier — inspect chained error, then retry the chunk/commit\n    }\n    r => r?,\n}","preventionTips":["Keep BE connections stable: check network MTU, idle timeouts on LBs between RW and BE","Size `stream_load_http_timeout` above worst-case chunk upload duration","Watch BE logs for stream-load aborts; retry the sink commit after transient failures"],"tags":["network","sink","doris","channel","stream-load"],"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"}