{"record":{"id":"798d6b5ce314aca9","repo":"vectordotdev/vector","slug":"consumer-reference-was-not-initialized","errorCode":null,"errorMessage":"Consumer reference was not initialized.","messagePattern":"Consumer reference was not initialized\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":1400,"sourceCode":"            .send(KafkaCallback::PartitionsRevoked(\n                tpl.elements()\n                    .iter()\n                    .map(|tp| (tp.topic().into(), tp.partition()))\n                    .collect(),\n                send,\n            ))\n            .ok();\n\n        while rendezvous.recv().is_ok() {\n            self.commit_consumer_state();\n        }\n    }\n\n    fn commit_consumer_state(&self) {\n        if let Some(consumer) = self\n            .consumer\n            .get()\n            .expect(\"Consumer reference was not initialized.\")\n            .upgrade()\n        {\n            match consumer.commit_consumer_state(CommitMode::Sync) {\n                Ok(_) | Err(KafkaError::ConsumerCommit(RDKafkaErrorCode::NoOffset)) => {\n                    /* Success, or nothing to do - yay \\0/ */\n                }\n                Err(error) => emit!(KafkaOffsetUpdateError { error }),\n            }\n        }\n    }\n}\n\nimpl ClientContext for KafkaSourceContext {\n    fn stats(&self, statistics: Statistics) {\n        self.stats.stats(statistics)\n    }\n}\n","sourceCodeStart":1382,"sourceCodeEnd":1418,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L1382-L1418","documentation":"Panic in the Kafka source's offset-commit path. `self.consumer` is a `OnceLock<Weak<StreamConsumer<KafkaSourceContext>>>` (src/sources/kafka.rs:1315) that is populated once the librdkafka consumer has been created during source startup. `commit_consumer_state` calls `.get().expect(\"Consumer reference was not initialized.\")`; if the commit loop (driven by a rendezvous channel that flushes offsets) runs before that `set()` call, `get()` returns None and the task panics. A set-but-dropped consumer is handled gracefully via `Weak::upgrade()`, so this panic strictly means the cell was never initialized.","triggerScenarios":"The offset-commit rendezvous channel delivers a message (periodic commit tick, rebalance, or shutdown flush) before `run()` has stored the consumer in the OnceLock; or the source is torn down while consumer construction (librdkafka client init) is still in flight so the commit task races ahead of initialization.","commonSituations":"Almost always an internal ordering bug or a custom build/test that calls commit_consumer_state directly. In released Vector versions the commit task is spawned after `consumer.set(...)`, so seeing this in production points to a patched/forked source or a version where startup ordering regressed.","solutions":["Upgrade Vector or patch src/sources/kafka.rs so `consumer.set(...)` happens strictly before the rendezvous commit loop is spawned","If embedding: replace the expect with `self.consumer.get().and_then(Weak::upgrade)` so an uninitialized or dead consumer skips the commit instead of panicking","Check earlier logs for a prior panic during consumer creation (e.g. librdkafka config error) that left the commit loop running without initialization"],"exampleFix":"// before\nlet Some(consumer) = self\n    .consumer\n    .get()\n    .expect(\"Consumer reference was not initialized.\")\n    .upgrade() else { return; };\n// after\nlet Some(consumer) = self.consumer.get().and_then(Weak::upgrade) else {\n    warn!(message = \"Skipping offset commit: consumer not yet initialized or already dropped.\");\n    return;\n};","handlingStrategy":"validation","validationCode":"// Rust (embedding): before starting the offset-commit loop\ndebug_assert!(source.consumer.get().is_some(),\n    \"consumer OnceLock must be set before commit task starts\");\nif source.consumer.get().is_none() {\n    return Err(\"kafka consumer not initialized; defer offset commits\".into());\n}","typeGuard":"fn consumer_initialized(source: &KafkaSource) -> bool {\n    source.consumer.get().is_some()\n}","tryCatchPattern":"// wrap the commit loop so one panic cannot kill the source silently\nlet result = std::panic::catch_unwind(AssertUnwindSafe(|| {\n    while rendezvous.recv().is_ok() {\n        source.commit_consumer_state();\n    }\n}));\nif result.is_err() {\n    error!(message = \"offset commit task panicked; continuing without commits\");\n}","preventionTips":["Keep `consumer.set(...)` strictly before spawning the commit loop in forks/patches","Treat any kafka source startup failure as fatal for the whole source, not just the consumer half","Test shutdown paths while consumer creation is delayed (bad broker list) to catch ordering races"],"tags":["kafka","oncelock","offset-commit","initialization","panic"],"backgroundTag":"missing-initialization","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}