{"record":{"id":"bd2f761efe3469df","repo":"risingwavelabs/risingwave","slug":"topic-not-found-on-broker-available-topics","errorCode":null,"errorMessage":"topic {} not found on broker, available topics: {:?}","messagePattern":"topic (.+?) not found on broker, available topics: (.+?)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/connector/src/source/pulsar/topic.rs","lineNumber":187,"sourceCode":"    }\n\n    Ok(parsed_topic)\n}\n\npub(crate) async fn check_topic_exists(\n    client: &Pulsar<TokioExecutor>,\n    topic: &Topic,\n) -> Result<()> {\n    // issue about api `get_topics_of_namespace`:\n    // for partitioned topic, the api will return all sub-topic of the topic instead of the topic itself\n    // Reduce async state machine size (see `clippy::large_futures`).\n    let topics_on_broker = Box::pin(client.get_topics_of_namespace(\n        format!(\"{}/{}\", topic.tenant, topic.namespace),\n        LookupMode::All,\n    ))\n    .await?;\n    if !topics_on_broker.contains(&topic.to_string()) {\n        bail!(\n            \"topic {} not found on broker, available topics: {:?}\",\n            topic,\n            topics_on_broker\n        );\n    }\n\n    Ok(())\n}\n\n#[cfg(test)]\nmod test {\n    use crate::source::pulsar::topic::{get_partition_index, parse_topic};\n\n    #[test]\n    fn test_parse_topic() {\n        assert_eq!(\n            parse_topic(\"success\").unwrap().to_string(),\n            \"persistent://public/default/success\".to_owned()","sourceCodeStart":169,"sourceCodeEnd":205,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/pulsar/topic.rs#L169-L205","documentation":"Before creating splits, the connector verifies that the configured topic actually exists on the Pulsar broker by listing all topics of the tenant/namespace. If the broker's topic list does not contain the configured topic, check_topic_exists bails with the full list of available topics to aid debugging.","triggerScenarios":"list_splits calling check_topic_exists when the topic was deleted, never created, or the tenant/namespace in the config does not match where the topic lives; also when broker-side topic auto-creation is disabled and the topic is only materialized on first produce.","commonSituations":"Typo in topic name in the CREATE SOURCE statement, topic deleted after source creation, connecting to the wrong broker/namespace (e.g. staging vs prod), or broker rejecting listing due to permissions so the list is empty/partial.","solutions":["Compare the configured topic against the 'available topics' list in the error and fix the name in the source definition","Create the topic on the broker (pulsar-admin topics create persistent://tenant/ns/topic) or enable auto-creation","Verify the tenant/namespace and broker service URL point to the intended cluster","Check broker permissions: the connector's credentials must be allowed to list topics in that namespace"],"exampleFix":"// before\nWITH connector = 'pulsar', topic = 'persistent://public/default/event'\n// after\nWITH connector = 'pulsar', topic = 'persistent://public/default/events'","handlingStrategy":"try-catch","validationCode":"// before creating the source, verify the topic exists\nawait admin.topics().listNamespaceTopics('tenant/namespace').then(topics => {\n  if (!topics.includes('persistent://tenant/namespace/mytopic')) throw new Error('topic missing');\n});","typeGuard":null,"tryCatchPattern":"match list_splits_result { Err(e) if e.contains(\"not found on broker\") => { verify_topic_name_and_broker(); retry_after_creating_topic(); } , r => r }","preventionTips":["Verify the topic exists with `pulsar-admin topics list` before CREATE SOURCE","Confirm tenant/namespace and broker service URL point to the intended cluster","Grant the connector's role permission to list topics in the namespace","Re-check after environment changes (topic deletion, cluster migration)"],"tags":["pulsar","network","resource-not-found"],"backgroundTag":"resource-not-found","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}