{"record":{"id":"65d74b1acb588e50","repo":"vectordotdev/vector","slug":"messages-are-consumed-in-dedicated-tasks-for-each","errorCode":null,"errorMessage":"Messages are consumed in dedicated tasks for each partition.","messagePattern":"Messages are consumed in dedicated tasks for each partition\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":968,"sourceCode":"        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,\n    finalizer: &Option<OrderedFinalizer<FinalizerEntry>>,\n    log_namespace: LogNamespace,\n) {","sourceCodeStart":950,"sourceCodeEnd":986,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L950-L986","documentation":"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.')","triggerScenarios":"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.","commonSituations":"rdkafka upgrades altering partition-queue forwarding behavior; races between assignment callbacks and queue splitting during rapid rebalances.","solutions":["Upgrade Vector - partition queue setup during rebalances was hardened across releases","Reduce rebalance frequency (assignor choice, session timeouts, staggered restarts)","Capture the offending message's topic/partition via debug logs and report","Validate against the same rdkafka version the release was built with"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"// Contain at task boundary and restart the consumer:\nif let Err(je) = tokio::spawn(kafka::run(cfg, shutdown)).await {\n    if je.is_panic() { /* log topic/partition context, restart with backoff */ }\n}","preventionTips":["Reduce rebalance frequency so queue-splitting races are rare","Match rdkafka versions to the tested release when building custom images","Report message-on-main-stream occurrences with broker and version details"],"tags":["rust","vector","kafka","rdkafka","partition-queue","panic","invariant"],"backgroundTag":"kafka-partition-consumer-invariant","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-08T15:18:49.778Z"}