{"record":{"id":"0c91fcbeed9b42de","repo":"vectordotdev/vector","slug":"messagestream-never-returns-ready-none","errorCode":null,"errorMessage":"MessageStream never returns Ready(None)","messagePattern":"MessageStream never returns Ready\\(None\\)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":965,"sourceCode":"    Handle::current().block_on(async move {\n        let mut eof: OptionFuture<_> = eof.into();\n        let mut stream = consumer.stream();\n        loop {\n            tokio::select! {\n                _ = &mut shutdown => {\n                    consumer.context().shutdown();\n                    break\n                },\n\n                Some(_) = &mut eof => {\n                    consumer.context().shutdown();\n                    break\n                },\n\n                // NB: messages are not received on this thread, however we poll\n                // the consumer to serve client callbacks, such as rebalance notifications\n                message = stream.next() => match message {\n                    None => unreachable!(\"MessageStream never returns Ready(None)\"),\n                    Some(Err(error)) => emit!(KafkaReadError { error }),\n                    Some(Ok(_msg)) => {\n                        unreachable!(\"Messages are consumed in dedicated tasks for each partition.\")\n                    }\n                },\n            }\n        }\n    });\n}\n\n#[allow(clippy::too_many_arguments)]\nasync fn parse_message(\n    msg: BorrowedMessage<'_>,\n    decoder: Decoder,\n    decompressor: Option<&Decompressor>,\n    keys: &'_ Keys,\n    out: &mut SourceSender,\n    acknowledgements: bool,","sourceCodeStart":947,"sourceCodeEnd":983,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L947-L983","documentation":"The main Kafka driver loop polls the StreamConsumer's message stream only to pump client callbacks; messages are consumed in dedicated per-partition tasks. The stream is contracted never to yield None, so that branch is unreachable - firing it means the rdkafka stream terminated, breaking the driver's core assumption.","triggerScenarios":"The rdkafka MessageStream ending (consumer context dropped beneath it, or a version where the stream terminates after error/shutdown) while the driver loop is still selecting on it.","commonSituations":"rdkafka/tokio-util version drift; embeddings that drop the consumer context while the source runs; librdkafka fatal errors ending the stream in newer client versions.","solutions":["Upgrade Vector so the rdkafka version matches the tested behavior","Verify the deployment doesn't swap librdkafka (ldd check inside the image)","Log with RUST_LOG=rdkafka=trace to see the terminal stream state before the panic","Report with versions - stream termination indicates an upstream contract change"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// The driver loop's panic ends the source; supervise and restart:\nif let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {\n    if je.is_panic() { /* log payload, restart with backoff; report upstream contract change */ }\n}","preventionTips":["Use official Vector builds so rdkafka/librdkafka stay version-matched","Don't drop the consumer context while the source task is running (embeddings)","Capture rdkafka=trace logs when stream-termination panics appear"],"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"}