risingwavelabs/risingwave · error

topic not found on broker, available topics

Error message

topic {} not found on broker, available topics: {:?}

What it means

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.

Solutions

1. Compare the configured topic against the 'available topics' list in the error and fix the name in the source definition
2. Create the topic on the broker (pulsar-admin topics create persistent://tenant/ns/topic) or enable auto-creation
3. Verify the tenant/namespace and broker service URL point to the intended cluster
4. Check broker permissions: the connector's credentials must be allowed to list topics in that namespace

Example fix

// before
WITH connector = 'pulsar', topic = 'persistent://public/default/event'
// after
WITH connector = 'pulsar', topic = 'persistent://public/default/events'

// before creating the source, verify the topic exists
await admin.topics().listNamespaceTopics('tenant/namespace').then(topics => {
  if (!topics.includes('persistent://tenant/namespace/mytopic')) throw new Error('topic missing');
});

match list_splits_result { Err(e) if e.contains("not found on broker") => { verify_topic_name_and_broker(); retry_after_creating_topic(); } , r => r }

When it happens

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

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

Understand the failure class

Background: 'Could not be found', 'does not exist', 'not found in database': the resource-not-found family when an ID, slug, key, or URI lookup comes back empty — this error's family across 20 libraries.

Related errors

• {connection_err (pulsar::Error) after retries exhausted}
• {pulsar::Error}
• Pulsar error
• {pulsar::Error from delivery future}
• all request confluent registry all timeout

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/bd2f761efe3469df. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/source/pulsar/topic.rs:187

    }

    Ok(parsed_topic)
}

pub(crate) async fn check_topic_exists(
    client: &Pulsar<TokioExecutor>,
    topic: &Topic,
) -> Result<()> {
    // issue about api `get_topics_of_namespace`:
    // for partitioned topic, the api will return all sub-topic of the topic instead of the topic itself
    // Reduce async state machine size (see `clippy::large_futures`).
    let topics_on_broker = Box::pin(client.get_topics_of_namespace(
        format!("{}/{}", topic.tenant, topic.namespace),
        LookupMode::All,
    ))
    .await?;
    if !topics_on_broker.contains(&topic.to_string()) {
        bail!(
            "topic {} not found on broker, available topics: {:?}",
            topic,
            topics_on_broker
        );
    }

    Ok(())
}

#[cfg(test)]
mod test {
    use crate::source::pulsar::topic::{get_partition_index, parse_topic};

    #[test]
    fn test_parse_topic() {
        assert_eq!(
            parse_topic("success").unwrap().to_string(),
            "persistent://public/default/success".to_owned()

View on GitHub (pinned to 6469eb736d)