{"record":{"id":"aa6c4a351ef2ec4d","repo":"quickwit-oss/quickwit","slug":"topic-does-not-exist","errorCode":null,"errorMessage":"topic `{}` does not exist","messagePattern":"topic `(.+?)` does not exist","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/kafka_source.rs","lineNumber":643,"sourceCode":"pub(super) async fn check_connectivity(params: KafkaSourceParams) -> anyhow::Result<()> {\n    let mut client_config = parse_client_params(params.client_params)?;\n\n    let consumer: BaseConsumer<DefaultConsumerContext> = client_config\n        .set(\"group.id\", \"quickwit-connectivity-check\".to_string())\n        .set_log_level(RDKafkaLogLevel::Error)\n        .create()?;\n\n    let topic = params.topic.clone();\n    let timeout = Timeout::After(Duration::from_secs(5));\n    let cluster_metadata = spawn_blocking(move || {\n        consumer\n            .fetch_metadata(Some(&topic), timeout)\n            .with_context(|| format!(\"failed to fetch metadata for topic `{topic}`\"))\n    })\n    .await??;\n\n    if cluster_metadata.topics().is_empty() {\n        bail!(\"topic `{}` does not exist\", params.topic);\n    }\n    let topic_metadata = &cluster_metadata.topics()[0];\n    assert_eq!(topic_metadata.name(), params.topic); // Belt and suspenders.\n\n    if topic_metadata.partitions().is_empty() {\n        bail!(\"topic `{}` has no partitions\", params.topic);\n    }\n    Ok(())\n}\n\n/// Creates a new `KafkaSourceConsumer`.\nfn create_consumer(\n    index_uid: &IndexUid,\n    source_id: &str,\n    params: KafkaSourceParams,\n    events_tx: mpsc::Sender<KafkaEvent>,\n) -> anyhow::Result<(ClientConfig, RdKafkaConsumer, GroupId)> {\n    // Group ID is limited to 255 characters.","sourceCodeStart":625,"sourceCodeEnd":661,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/kafka_source.rs#L625-L661","documentation":"During Kafka connectivity checking, metadata was fetched for the topic but the returned cluster metadata contained no topics, so Quickwit concludes the topic does not exist and the source cannot consume from it.","triggerScenarios":"`check_connectivity` fetches metadata via `fetch_metadata(Some(&topic), timeout)`; the `rdkafka` response comes back with an empty `topics()` list. (Note: a real `Err` from fetch_metadata would instead surface the 'failed to fetch metadata' context error.)","commonSituations":"Topic deleted after the source was configured; typo in the topic name in the source config; broker auto-create disabled and topic never created; connecting to the wrong Kafka cluster.","solutions":["Verify the topic exists: `kafka-topics.sh --bootstrap-server <broker> --describe --topic <topic>` and fix the topic name in the source config.","Create the missing topic if it was deleted.","Confirm the bootstrap_servers/client_params in the source config point at the intended cluster.","Re-run the connectivity check: `quickwit source check-connectivity`."],"exampleFix":"// before\nsource:\n  params:\n    topic: log-events-prod\n// after (topic actually named differently)\nkafka-topics.sh --bootstrap-server broker:9092 --list\nsource:\n  params:\n    topic: log-events","handlingStrategy":"validation","validationCode":"kafka-topics.sh --bootstrap-server $BROKERS --describe --topic $TOPIC || echo \"topic $TOPIC missing\"","typeGuard":null,"tryCatchPattern":"// run connectivity check before enabling the source\nmatch quickwit_source_check(topic, brokers).await {\n    Err(e) if e.to_string().contains(\"does not exist\") => create_or_fix_topic().await?,\n    other => other?,\n}","preventionTips":["Create the topic with explicit partitions/replication before configuring the source.","Keep topic names in source configs in sync with IaC (Terraform/Strimzi).","Run `quickwit source check-connectivity` after every config change."],"tags":["kafka","topic-not-found","configuration","rust"],"backgroundTag":"resource-not-found","analyzedSha":"a39730c5cdcd1a4fe798403737ae293999ea21f8","analyzedAt":"2026-09-08T13:19:37.784Z","contentChangedAt":"2026-09-08T13:19:37.784Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}