{"record":{"id":"ec75adb1876482e1","repo":"risingwavelabs/risingwave","slug":"broken-hash-shuffle-channel","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/consistent_hash_shuffle_channel.rs","lineNumber":153,"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 ConsistentHashShuffleReceiver {\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_consistent_shuffle_channel(\n    shuffle: &ExchangeInfo,\n    output_channel_size: usize,\n) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {\n    let consistent_hash_info = match shuffle.distribution {\n        Some(exchange_info::Distribution::ConsistentHashInfo(ref v)) => v.clone(),\n        _ => exchange_info::ConsistentHashInfo::default(),\n    };\n\n    let output_count = consistent_hash_info\n        .vmap\n        .iter()\n        .copied()\n        .sorted()","sourceCodeStart":135,"sourceCodeEnd":171,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/src/task/consistent_hash_shuffle_channel.rs#L135-L171","documentation":"The consistent-hash shuffle channel used by batch exchange was closed by its sender(s) before a normal termination signal arrived. `ConsistentHashShuffleReceiver::recv` treats `None` from the underlying mpsc receiver (all sender handles dropped) as an abnormal condition — a healthy shuffle always ends with an explicit Done/barrier — and surfaces it as the `Internal` error \"broken hash_shuffle_channel\".","triggerScenarios":"Raised inside `ConsistentHashShuffleReceiver::recv` (src/batch/src/task/consistent_hash_shuffle_channel.rs:153) when `recv().await` on the mpsc receiver returns `None`, i.e. all `ChanSender` clones for this shuffle (created via `new_consistent_shuffle_channel`) were dropped before sending the final barrier/Done chunk.","commonSituations":"A distributed query where one upstream sender task panicked or failed (e.g. RPC error fetching data from another node) and dropped its sender; the stage was cancelled mid-shuffle; a producer executor bug exits its loop early without sending Done to all shuffled receivers.","solutions":["Look for the root-cause failure in the sender/upstream task logs; this error is downstream collateral of an earlier failure.","Ensure every shuffled producer sends its final barrier/Done chunk on all paths, including error paths, before dropping senders.","Check cluster node health and network stability between compute nodes when this appears during distributed queries.","Avoid using sender drop as the completion mechanism; always signal termination explicitly through the channel."],"exampleFix":"// before: early exit from shuffle producer without sending Done to all partitions\nfor chunk in chunks {\n    if let Err(e) = process(chunk) {\n        return Err(e); // senders dropped -> \"broken hash_shuffle_channel\" downstream\n    }\n}\n// after: signal all partitions before exiting\nfor chunk in chunks {\n    if let Err(e) = process(chunk) {\n        for sender in &mut senders {\n            sender.send(Err(e.clone())).await?;\n        }\n        return Err(e);\n    }\n}","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"match receiver.recv().await {\n    Ok(Some(chunk)) => { /* process */ }\n    Ok(None) => { /* done */ }\n    Err(e) if e.to_string().contains(\"broken\") => {\n        return Err(anyhow!(\"shuffle sender failed; check upstream task error\"))\n            .context(e.to_string());\n    }\n    Err(e) => return Err(e.into()),\n}","preventionTips":["In shuffle producers, send Done/barrier to every partition before dropping any sender","Propagate producer errors through the channel as Err chunks","Verify network/RPC stability between nodes for distributed shuffle queries","Review producer executor loops for early `?` returns that skip the termination signal"],"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"}