vectordotdev/vector · error

Consumer reference was not initialized.

Error message

Consumer reference was not initialized.

What it means

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.

Source

Thrown at src/sources/kafka.rs:1400

            .send(KafkaCallback::PartitionsRevoked(
                tpl.elements()
                    .iter()
                    .map(|tp| (tp.topic().into(), tp.partition()))
                    .collect(),
                send,
            ))
            .ok();

        while rendezvous.recv().is_ok() {
            self.commit_consumer_state();
        }
    }

    fn commit_consumer_state(&self) {
        if let Some(consumer) = self
            .consumer
            .get()
            .expect("Consumer reference was not initialized.")
            .upgrade()
        {
            match consumer.commit_consumer_state(CommitMode::Sync) {
                Ok(_) | Err(KafkaError::ConsumerCommit(RDKafkaErrorCode::NoOffset)) => {
                    /* Success, or nothing to do - yay \0/ */
                }
                Err(error) => emit!(KafkaOffsetUpdateError { error }),
            }
        }
    }
}

impl ClientContext for KafkaSourceContext {
    fn stats(&self, statistics: Statistics) {
        self.stats.stats(statistics)
    }
}

View on GitHub (pinned to 3708c39b12)

Solutions

1. Upgrade Vector or patch src/sources/kafka.rs so `consumer.set(...)` happens strictly before the rendezvous commit loop is spawned
2. 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
3. Check earlier logs for a prior panic during consumer creation (e.g. librdkafka config error) that left the commit loop running without initialization

Example fix

// before
let Some(consumer) = self
    .consumer
    .get()
    .expect("Consumer reference was not initialized.")
    .upgrade() else { return; };
// after
let Some(consumer) = self.consumer.get().and_then(Weak::upgrade) else {
    warn!(message = "Skipping offset commit: consumer not yet initialized or already dropped.");
    return;
};

// Rust (embedding): before starting the offset-commit loop
debug_assert!(source.consumer.get().is_some(),
    "consumer OnceLock must be set before commit task starts");
if source.consumer.get().is_none() {
    return Err("kafka consumer not initialized; defer offset commits".into());
}

fn consumer_initialized(source: &KafkaSource) -> bool {
    source.consumer.get().is_some()
}

// wrap the commit loop so one panic cannot kill the source silently
let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
    while rendezvous.recv().is_ok() {
        source.commit_consumer_state();
    }
}));
if result.is_err() {
    error!(message = "offset commit task panicked; continuing without commits");
}

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• Error setting up consumer context.
• double chunk_size_events initialization
• double thread initialization
• MessageStream never calls Ready(None)
• Partition consumer finished after completion.

AI-assisted analysis of vectordotdev/vector@3708c39b12 (2026-08-20). Data as JSON: /api/errors/798d6b5ce314aca9. Report an issue: GitHub.