risingwavelabs/risingwave · error · BatchError

broken hash_shuffle_channel

Error message

broken hash_shuffle_channel

What it means

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".

Source

Thrown at src/batch/src/task/consistent_hash_shuffle_channel.rs:153

    async fn send_done(self, error: Option<Arc<BatchError>>) -> BatchResult<()> {
        for sender in self.senders {
            sender
                .send(error.clone().map(Err).unwrap_or(Ok(None)))
                .await
                .map_err(|_| SenderError)?
        }

        Ok(())
    }
}

impl ChanReceiver for ConsistentHashShuffleReceiver {
    async fn recv(&mut self) -> SharedResult<Option<DataChunkInChannel>> {
        match self.receiver.recv().await {
            Some(data_chunk) => data_chunk,
            // Early close should be treated as error.
            None => Err(Arc::new(Internal(anyhow!("broken hash_shuffle_channel")))),
        }
    }
}

pub fn new_consistent_shuffle_channel(
    shuffle: &ExchangeInfo,
    output_channel_size: usize,
) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {
    let consistent_hash_info = match shuffle.distribution {
        Some(exchange_info::Distribution::ConsistentHashInfo(ref v)) => v.clone(),
        _ => exchange_info::ConsistentHashInfo::default(),
    };

    let output_count = consistent_hash_info
        .vmap
        .iter()
        .copied()
        .sorted()

View on GitHub (pinned to 6469eb736d)

Solutions

1. Look for the root-cause failure in the sender/upstream task logs; this error is downstream collateral of an earlier failure.
2. Ensure every shuffled producer sends its final barrier/Done chunk on all paths, including error paths, before dropping senders.
3. Check cluster node health and network stability between compute nodes when this appears during distributed queries.
4. Avoid using sender drop as the completion mechanism; always signal termination explicitly through the channel.

Example fix

// before: early exit from shuffle producer without sending Done to all partitions
for chunk in chunks {
    if let Err(e) = process(chunk) {
        return Err(e); // senders dropped -> "broken hash_shuffle_channel" downstream
    }
}
// after: signal all partitions before exiting
for chunk in chunks {
    if let Err(e) = process(chunk) {
        for sender in &mut senders {
            sender.send(Err(e.clone())).await?;
        }
        return Err(e);
    }
}

match receiver.recv().await {
    Ok(Some(chunk)) => { /* process */ }
    Ok(None) => { /* done */ }
    Err(e) if e.to_string().contains("broken") => {
        return Err(anyhow!("shuffle sender failed; check upstream task error"))
            .context(e.to_string());
    }
    Err(e) => return Err(e.into()),
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• broken hash_shuffle_channel
• broken broadcast_channel
• broken fifo_channel
• query_epoch not set in distributed lookup join
• Storage error: {0}

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