vectordotdev/vector · error

Messages are consumed in dedicated tasks for each partition.

Error message

Messages are consumed in dedicated tasks for each partition.

What it means

The driver loop's comment states messages are not received on that thread - polling stream.next() serves only to serve rdkafka callbacks. An Ok(item) means an actual message was delivered on the main consumer stream, which cannot happen once every partition is consumed through split partition queues; hence unreachable!('Messages are consumed in dedicated tasks for each partition.')

Source

Thrown at src/sources/kafka.rs:968

        loop {
            tokio::select! {
                _ = &mut shutdown => {
                    consumer.context().shutdown();
                    break
                },

                Some(_) = &mut eof => {
                    consumer.context().shutdown();
                    break
                },

                // NB: messages are not received on this thread, however we poll
                // the consumer to serve client callbacks, such as rebalance notifications
                message = stream.next() => match message {
                    None => unreachable!("MessageStream never returns Ready(None)"),
                    Some(Err(error)) => emit!(KafkaReadError { error }),
                    Some(Ok(_msg)) => {
                        unreachable!("Messages are consumed in dedicated tasks for each partition.")
                    }
                },
            }
        }
    });
}

#[allow(clippy::too_many_arguments)]
async fn parse_message(
    msg: BorrowedMessage<'_>,
    decoder: Decoder,
    decompressor: Option<&Decompressor>,
    keys: &'_ Keys,
    out: &mut SourceSender,
    acknowledgements: bool,
    finalizer: &Option<OrderedFinalizer<FinalizerEntry>>,
    log_namespace: LogNamespace,
) {

View on GitHub (pinned to 3708c39b12)

Solutions

1. Upgrade Vector - partition queue setup during rebalances was hardened across releases
2. Reduce rebalance frequency (assignor choice, session timeouts, staggered restarts)
3. Capture the offending message's topic/partition via debug logs and report
4. Validate against the same rdkafka version the release was built with

// Contain at task boundary and restart the consumer:
if let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {
    if je.is_panic() { /* log topic/partition context, restart with backoff */ }
}

When it happens

Trigger: A message arriving on the un-split consumer stream: a partition whose queue wasn't yet split being polled on the main stream, or an rdkafka version changing split_partition_queue/forwarding semantics so messages leak past the partition queues.

Common situations: rdkafka upgrades altering partition-queue forwarding behavior; races between assignment callbacks and queue splitting during rapid rebalances.

Related errors

• MessageStream never calls Ready(None)
• MessageStream never returns Ready(None)
• Partition consumer finished after completion.
• Partition assignment received after completion.
• Partitions revoked after completion.

AI-assisted analysis of vectordotdev/vector@3708c39b12 (2026-08-20). Data as JSON: /api/errors/65d74b1acb588e50. Report an issue: GitHub.