vectordotdev/vector · error

MessageStream never calls Ready(None)

Error message

MessageStream never calls Ready(None)

What it means

In the Kafka source, each partition is consumed by a dedicated task whose tokio::select! polls rdkafka's MessageStream. That stream is contracted to never terminate (librdkafka reports errors as items and keeps polling), so the None branch is marked unreachable. If the underlying stream ever yields Ready(None) the invariant is broken and the partition task panics.

Source

Thrown at src/sources/kafka.rs:661

                    ack = ack_stream.next() => match ack {
                        Some((status, entry)) => {
                            if status == BatchStatus::Delivered
                                && let Err(error) =  consumer.store_offset(&entry.topic, entry.partition, entry.offset) {
                                    emit!(KafkaOffsetUpdateError { error });
                                }
                        }
                        None if finalizer.is_none() => {
                            debug!("Acknowledgement stream complete for partition {}:{}.", &tp.0, tp.1);
                            break
                        }
                        None => {
                            debug!("Acknowledgement stream empty for {}:{}", &tp.0, tp.1);
                        }
                    },

                    message = messages.next(), if finalizer.is_some() => match message {
                        None => unreachable!("MessageStream never calls Ready(None)"),
                        Some(Err(error)) => match error {
                            rdkafka::error::KafkaError::PartitionEOF(partition) if exit_eof => {
                                debug!("EOF for partition {}.", partition);
                                status = PartitionConsumerStatus::PartitionEOF;
                                finalizer.take();
                            },
                            _ => emit!(KafkaReadError { error }),
                        },
                        Some(Ok(msg)) => {
                            emit!(KafkaBytesReceived {
                                byte_size: msg.payload_len(),
                                protocol: "tcp",
                                topic: msg.topic(),
                                partition: msg.partition(),
                            });
                            parse_message(msg, decoder.clone(), decompressor.as_ref(), &keys, &mut out, acknowledgements, &finalizer, log_namespace).await;
                        }
                    },

View on GitHub (pinned to 3708c39b12)

Solutions

1. Upgrade Vector to a release pinned to a tested rdkafka/librdkafka pair
2. Verify exactly one librdkafka is linked and of the expected major version: ldd $(which vector) | grep rdkafka
3. Reproduce with RUST_LOG=vector=debug,rdkafka=trace and capture the last stream items before the panic
4. Open an issue with the rdkafka/librdkafka versions - the stream contract was violated upstream

// Partition tasks already isolate the panic to one task; the topology then fails.
// Supervise at the process boundary and keep the backtrace:
//   systemd: Restart=on-failure, Environment=RUST_BACKTRACE=1
let h = tokio::spawn(consume_partition(topic_partition, ...));
if let Err(je) = h.await {
    if je.is_panic() { /* log partition + backtrace, restart the source with backoff */ }
}

When it happens

Trigger: An rdkafka/tokio version combination where the StreamConsumer poll stream can end (client shutdown or context cancellation surfaced as stream end); a librdkafka behavior change across versions; the consumer context dropped mid-poll while the select loop still runs.

Common situations: Upgrading rdkafka or swapping the system librdkafka under a Vector build expecting the bundled one; brokers/proxies triggering client-level fatal errors that end the stream; custom builds with mismatched rdkafka features.

Related errors

• MessageStream never returns Ready(None)
• Messages are consumed in dedicated tasks for each partition.
• 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/33b8b2345204a23f. Report an issue: GitHub.