{"record":{"id":"33b8b2345204a23f","repo":"vectordotdev/vector","slug":"messagestream-never-calls-ready-none","errorCode":null,"errorMessage":"MessageStream never calls Ready(None)","messagePattern":"MessageStream never calls Ready\\(None\\)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":661,"sourceCode":"\n                    ack = ack_stream.next() => match ack {\n                        Some((status, entry)) => {\n                            if status == BatchStatus::Delivered\n                                && let Err(error) =  consumer.store_offset(&entry.topic, entry.partition, entry.offset) {\n                                    emit!(KafkaOffsetUpdateError { error });\n                                }\n                        }\n                        None if finalizer.is_none() => {\n                            debug!(\"Acknowledgement stream complete for partition {}:{}.\", &tp.0, tp.1);\n                            break\n                        }\n                        None => {\n                            debug!(\"Acknowledgement stream empty for {}:{}\", &tp.0, tp.1);\n                        }\n                    },\n\n                    message = messages.next(), if finalizer.is_some() => match message {\n                        None => unreachable!(\"MessageStream never calls Ready(None)\"),\n                        Some(Err(error)) => match error {\n                            rdkafka::error::KafkaError::PartitionEOF(partition) if exit_eof => {\n                                debug!(\"EOF for partition {}.\", partition);\n                                status = PartitionConsumerStatus::PartitionEOF;\n                                finalizer.take();\n                            },\n                            _ => emit!(KafkaReadError { error }),\n                        },\n                        Some(Ok(msg)) => {\n                            emit!(KafkaBytesReceived {\n                                byte_size: msg.payload_len(),\n                                protocol: \"tcp\",\n                                topic: msg.topic(),\n                                partition: msg.partition(),\n                            });\n                            parse_message(msg, decoder.clone(), decompressor.as_ref(), &keys, &mut out, acknowledgements, &finalizer, log_namespace).await;\n                        }\n                    },","sourceCodeStart":643,"sourceCodeEnd":679,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L643-L679","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Upgrade Vector to a release pinned to a tested rdkafka/librdkafka pair","Verify exactly one librdkafka is linked and of the expected major version: ldd $(which vector) | grep rdkafka","Reproduce with RUST_LOG=vector=debug,rdkafka=trace and capture the last stream items before the panic","Open an issue with the rdkafka/librdkafka versions - the stream contract was violated upstream"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Partition tasks already isolate the panic to one task; the topology then fails.\n// Supervise at the process boundary and keep the backtrace:\n//   systemd: Restart=on-failure, Environment=RUST_BACKTRACE=1\nlet h = tokio::spawn(consume_partition(topic_partition, ...));\nif let Err(je) = h.await {\n    if je.is_panic() { /* log partition + backtrace, restart the source with backoff */ }\n}","preventionTips":["Pin Vector releases; don't swap system librdkafka versions beneath them","Check ldd output for mismatched or duplicate librdkafka in container images","Deploy with RUST_BACKTRACE=1 so invariant panics are actionable"],"tags":["rust","vector","kafka","rdkafka","source","stream","panic","invariant"],"backgroundTag":"rdkafka-stream-terminated","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T10:18:20.063Z"}