risingwavelabs/risingwave · error · BatchError

broken hash_shuffle_channel

Error message

broken hash_shuffle_channel

What it means

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

Source

Thrown at src/batch/src/task/hash_shuffle_channel.rs:149

    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 HashShuffleReceiver {
    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_hash_shuffle_channel(
    shuffle: &ExchangeInfo,
    output_channel_size: usize,
) -> (ChanSenderImpl, Vec<ChanReceiverImpl>) {
    let hash_info = match shuffle.distribution {
        Some(exchange_info::Distribution::HashInfo(ref v)) => v.clone(),
        _ => exchange_info::HashInfo::default(),
    };

    let output_count = hash_info.output_count as usize;
    let mut senders = Vec::with_capacity(output_count);
    let mut receivers = Vec::with_capacity(output_count);
    for _ in 0..output_count {
        let (s, r) = mpsc::channel(output_channel_size);

View on GitHub (pinned to 6469eb736d)

Solutions

1. Find and fix the upstream sender failure in the logs; this error is a downstream symptom of an earlier producer failure or cancellation.
2. Guarantee each partition's producer sends its final barrier/Done chunk before dropping the sender on all code paths.
3. Check inter-node network health and exchange timeouts if this occurs only in distributed runs.
4. Never rely on channel close as the completion signal in shuffle producers.

Example fix

// before: partition loop exits early, leaving some senders dropped without Done
for (i, sender) in senders.iter_mut().enumerate() {
    if let Err(e) = send_partition(sender, i).await {
        return Err(e); // remaining senders dropped -> broken hash_shuffle_channel
    }
}
// after: always complete the protocol for every partition
for (i, sender) in senders.iter_mut().enumerate() {
    if let Err(e) = send_partition(sender, i).await {
        sender.send(Err(e)).await?;
    } else {
        sender.send(Ok(done_chunk())).await?;
    }
}

match receiver.recv().await {
    Ok(Some(chunk)) => { /* process */ }
    Ok(None) => return Err(anyhow!("hash shuffle closed prematurely")),
    Err(e) => return Err(e), // upstream error already propagated as SharedResult::Err
}

When it happens

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

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

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/be292eb067c26156. Report an issue: GitHub.