{"record":{"id":"c0b1e63f2b89fb0e","repo":"vectordotdev/vector","slug":"error-setting-up-consumer-context","errorCode":null,"errorMessage":"Error setting up consumer context.","messagePattern":"Error setting up consumer context\\.","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/sources/kafka.rs","lineNumber":452,"sourceCode":"async fn kafka_source(\n    config: KafkaSourceConfig,\n    consumer: StreamConsumer<KafkaSourceContext>,\n    callback_rx: UnboundedReceiver<KafkaCallback>,\n    decoder: Decoder,\n    decompressor: Option<Decompressor>,\n    out: SourceSender,\n    shutdown: ShutdownSignal,\n    eof: bool,\n    log_namespace: LogNamespace,\n) -> Result<(), ()> {\n    let span = info_span!(\"kafka_source\");\n    let consumer = Arc::new(consumer);\n\n    consumer\n        .context()\n        .consumer\n        .set(Arc::downgrade(&consumer))\n        .expect(\"Error setting up consumer context.\");\n\n    // EOF signal allowing the coordination task to tell the kafka client task when all partitions have reached EOF\n    let (eof_tx, eof_rx) = eof.then(oneshot::channel::<()>).unzip();\n\n    let topics: Vec<&str> = config.topics.iter().map(|s| s.as_str()).collect();\n    if let Err(e) = consumer.subscribe(&topics).context(SubscribeSnafu) {\n        error!(\"{}\", e);\n        return Err(());\n    }\n\n    let coordination_task = {\n        let span = span.clone();\n        let consumer = Arc::clone(&consumer);\n        let drain_timeout_ms = config\n            .drain_timeout_ms\n            .map_or(config.session_timeout_ms / 2, Duration::from_millis);\n        let consumer_state = ConsumerStateInner::<Consuming>::new(\n            config,","sourceCodeStart":434,"sourceCodeEnd":470,"githubUrl":"https://github.com/vectordotdev/vector/blob/3708c39b12a93212ed8b8d7510b4cc7769cb5864/src/sources/kafka.rs#L434-L470","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Build a new StreamConsumer for every kafka_source invocation; never reuse one across restarts","If the consumer is already fresh per run, report it as a regression with config and Vector version","Patch: treat set() failure as benign (log a warning) since the slot then already holds the right reference"],"exampleFix":"// before\nconsumer\n    .context()\n    .consumer\n    .set(Arc::downgrade(&consumer))\n    .expect(\"Error setting up consumer context.\");\n\n// after\nif consumer\n    .context()\n    .consumer\n    .set(Arc::downgrade(&consumer))\n    .is_err()\n{\n    warn!(message = \"kafka consumer context already set; skipping re-init\");\n}","handlingStrategy":"validation","validationCode":"// build a fresh consumer for every source task and verify the slot is empty first:\nfn init_consumer_context(\n    consumer: &StreamConsumer<KafkaSourceContext>,\n) -> Result<(), KafkaSourceContextError> {\n    consumer\n        .context()\n        .consumer\n        .set(Arc::downgrade(consumer))\n        .map_err(|_| KafkaSourceContextError::AlreadyInitialized)\n}\n\n// always call with a consumer constructed in the same invocation","typeGuard":"fn context_is_unset(consumer: &StreamConsumer<KafkaSourceContext>) -> bool {\n    consumer.context().consumer.get().is_none()\n}","tryCatchPattern":"if consumer\n    .context()\n    .consumer\n    .set(Arc::downgrade(&consumer))\n    .is_err()\n{\n    warn!(message = \"kafka consumer context already set; skipping re-init\");\n}","preventionTips":["Construct a new StreamConsumer per kafka_source run; never share or reuse one across restarts","Treat set() failure as idempotent success with a warning, since the slot already holds the right weak reference","Add a regression test that starts the kafka source twice in one process"],"tags":["rust","panic","rdkafka","kafka","initialization"],"backgroundTag":"double-initialization","analyzedSha":"3708c39b12a93212ed8b8d7510b4cc7769cb5864","analyzedAt":"2026-08-20T07:02:18.786Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}