{"record":{"id":"be292eb067c26156","repo":"risingwavelabs/risingwave","slug":"broken-hash-shuffle-channel-be292e","errorCode":null,"errorMessage":"broken hash_shuffle_channel","messagePattern":"broken hash_shuffle_channel","errorType":"error_code","errorClass":"BatchError","httpStatus":null,"severity":"error","filePath":"src/batch/src/task/hash_shuffle_channel.rs","lineNumber":149,"sourceCode":"\n    async fn send_done(self, error: Option<Arc<BatchError>>) -> BatchResult<()> {\n        for sender in self.senders {\n            sender\n                .send(error.clone().map(Err).unwrap_or(Ok(None)))\n                .await\n                .map_err(|_| SenderError)?\n        }\n\n        Ok(())\n    }\n}\n\nimpl ChanReceiver for HashShuffleReceiver {\n    async fn recv(&mut self) -> SharedResult<Option<DataChunkInChannel>> {\n        match self.receiver.recv().await {\n            Some(data_chunk) => data_chunk,\n            // Early close should be treated as error.\n            None => Err(Arc::new(Internal(anyhow!(\"broken hash_shuffle_channel\")))),\n        }\n    }\n}\n\npub fn new_hash_shuffle_channel(\n    shuffle: &ExchangeInfo,\n    output_channel_size: usize,\n) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {\n    let hash_info = match shuffle.distribution {\n        Some(exchange_info::Distribution::HashInfo(ref v)) => v.clone(),\n        _ => exchange_info::HashInfo::default(),\n    };\n\n    let output_count = hash_info.output_count as usize;\n    let mut senders = Vec::with_capacity(output_count);\n    let mut receivers = Vec::with_capacity(output_count);\n    for _ in 0..output_count {\n        let (s, r) = mpsc::channel(output_channel_size);","sourceCodeStart":131,"sourceCodeEnd":167,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/src/task/hash_shuffle_channel.rs#L131-L167","documentation":"A hash-shuffle exchange channel was closed by all of its senders before a normal end-of-stream signal arrived. `HashShuffleReceiver::recv` interprets `None` from the underlying mpsc receiver (senders dropped without sending the final barrier/Done) as an abnormal teardown and raises the `Internal` error \"broken hash_shuffle_channel\".","triggerScenarios":"Raised inside `HashShuffleReceiver::recv` (src/batch/src/task/hash_shuffle_channel.rs:149) when the mpsc receiver created by `new_hash_shuffle_channel(shuffle, ...)` yields `None`, i.e. every sender clone was dropped before the terminal barrier chunk was sent.","commonSituations":"Upstream shuffle producers fail mid-query (RPC errors between nodes, executor panic, OOM) and drop their senders; the batch task is cancelled while shuffling; producer code paths that skip sending Done to some partitions (e.g. uneven hash partitions or early `?` returns).","solutions":["Find and fix the upstream sender failure in the logs; this error is a downstream symptom of an earlier producer failure or cancellation.","Guarantee each partition's producer sends its final barrier/Done chunk before dropping the sender on all code paths.","Check inter-node network health and exchange timeouts if this occurs only in distributed runs.","Never rely on channel close as the completion signal in shuffle producers."],"exampleFix":"// before: partition loop exits early, leaving some senders dropped without Done\nfor (i, sender) in senders.iter_mut().enumerate() {\n    if let Err(e) = send_partition(sender, i).await {\n        return Err(e); // remaining senders dropped -> broken hash_shuffle_channel\n    }\n}\n// after: always complete the protocol for every partition\nfor (i, sender) in senders.iter_mut().enumerate() {\n    if let Err(e) = send_partition(sender, i).await {\n        sender.send(Err(e)).await?;\n    } else {\n        sender.send(Ok(done_chunk())).await?;\n    }\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match receiver.recv().await {\n    Ok(Some(chunk)) => { /* process */ }\n    Ok(None) => return Err(anyhow!(\"hash shuffle closed prematurely\")),\n    Err(e) => return Err(e), // upstream error already propagated as SharedResult::Err\n}","preventionTips":["Ensure all hash partitions receive their terminal barrier, not just the ones with data","Investigate the first error in the log — this error is downstream of a producer failure","Avoid panic in producer executors; convert failures to channel errors","For flaky distributed runs, check node health and exchange timeouts"],"tags":["batch","shuffle","channel"],"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"}