vectordotdev/vector · error

Error setting up consumer context.

Error message

Error setting up consumer context.

What it means

The kafka source wraps rdkafka's StreamConsumer<KafkaSourceContext> in an Arc and stores a weak self-reference so client callbacks can reach the consumer: consumer.context().consumer.set(Arc::downgrade(&consumer)).expect("Error setting up consumer context."). The context slot's set() returns Err(previous value) when it is already occupied, so this panics exactly when this initialization runs twice for the same consumer instance.

Source

Thrown at src/sources/kafka.rs:452

async fn kafka_source(
    config: KafkaSourceConfig,
    consumer: StreamConsumer<KafkaSourceContext>,
    callback_rx: UnboundedReceiver<KafkaCallback>,
    decoder: Decoder,
    decompressor: Option<Decompressor>,
    out: SourceSender,
    shutdown: ShutdownSignal,
    eof: bool,
    log_namespace: LogNamespace,
) -> Result<(), ()> {
    let span = info_span!("kafka_source");
    let consumer = Arc::new(consumer);

    consumer
        .context()
        .consumer
        .set(Arc::downgrade(&consumer))
        .expect("Error setting up consumer context.");

    // EOF signal allowing the coordination task to tell the kafka client task when all partitions have reached EOF
    let (eof_tx, eof_rx) = eof.then(oneshot::channel::<()>).unzip();

    let topics: Vec<&str> = config.topics.iter().map(|s| s.as_str()).collect();
    if let Err(e) = consumer.subscribe(&topics).context(SubscribeSnafu) {
        error!("{}", e);
        return Err(());
    }

    let coordination_task = {
        let span = span.clone();
        let consumer = Arc::clone(&consumer);
        let drain_timeout_ms = config
            .drain_timeout_ms
            .map_or(config.session_timeout_ms / 2, Duration::from_millis);
        let consumer_state = ConsumerStateInner::<Consuming>::new(
            config,

View on GitHub (pinned to 3708c39b12)

Solutions

1. Build a new StreamConsumer for every kafka_source invocation; never reuse one across restarts
2. If the consumer is already fresh per run, report it as a regression with config and Vector version
3. Patch: treat set() failure as benign (log a warning) since the slot then already holds the right reference

Example fix

// before
consumer
    .context()
    .consumer
    .set(Arc::downgrade(&consumer))
    .expect("Error setting up consumer context.");

// after
if consumer
    .context()
    .consumer
    .set(Arc::downgrade(&consumer))
    .is_err()
{
    warn!(message = "kafka consumer context already set; skipping re-init");
}

// build a fresh consumer for every source task and verify the slot is empty first:
fn init_consumer_context(
    consumer: &StreamConsumer<KafkaSourceContext>,
) -> Result<(), KafkaSourceContextError> {
    consumer
        .context()
        .consumer
        .set(Arc::downgrade(consumer))
        .map_err(|_| KafkaSourceContextError::AlreadyInitialized)
}

// always call with a consumer constructed in the same invocation

fn context_is_unset(consumer: &StreamConsumer<KafkaSourceContext>) -> bool {
    consumer.context().consumer.get().is_none()
}

if consumer
    .context()
    .consumer
    .set(Arc::downgrade(&consumer))
    .is_err()
{
    warn!(message = "kafka consumer context already set; skipping re-init");
}

When it happens

Trigger: Calling kafka_source() (or the run wrapper that initializes the consumer) twice with the same StreamConsumer - re-running setup for a restart, sharing one consumer across tasks, or a regression that re-enters this function. A consumer freshly built per source run never trips it.

Common situations: Custom forks wiring consumer reuse or restarts; tests that call the source task with a cached consumer; stock Vector builds a new consumer per run, so hitting it there indicates a regression to report.

Related errors

• MessageStream never calls Ready(None)
• Partition consumer finished after completion.
• Partition assignment received after completion.
• Partitions revoked after completion.
• Shutdown received after completion.

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