risingwavelabs/risingwave · error

topic {} not found

Error message

topic {} not found

What it means

Raised when Kafka metadata for the configured topic cannot be resolved into exactly one topic entry. This bail fires in `fetch_topic_partition` when the metadata response does not contain exactly one topic (e.g. the topic was deleted, or the broker returned an error/unknown-topic-response), so partition enumeration cannot proceed.

Source

Thrown at src/connector/src/source/kafka/enumerator.rs:628

    pub async fn check_reachability(&self) -> ConnectorResult<()> {
        let _ = self
            .client
            .fetch_metadata(Some(self.topic.as_str()), self.sync_call_timeout)
            .await?;
        Ok(())
    }

    async fn fetch_topic_partition(&self) -> ConnectorResult<Vec<i32>> {
        // for now, we only support one topic
        let metadata = self
            .client
            .fetch_metadata(Some(self.topic.as_str()), self.sync_call_timeout)
            .await?;

        let topic_meta = match metadata.topics() {
            [meta] => meta,
            _ => bail!("topic {} not found", self.topic),
        };

        if topic_meta.partitions().is_empty() {
            bail!("topic {} not found", self.topic);
        }

        Ok(topic_meta
            .partitions()
            .iter()
            .map(|partition| partition.id())
            .collect())
    }
}

/// Drops cached start offsets of partitions that are no longer present and returns the
/// partitions whose start offset still has to be resolved.
fn sync_resolved_partitions(
    resolved: &mut HashMap<i32, Option<i64>>,

View on GitHub (pinned to 6469eb736d)

Solutions

1. Verify the topic exists: `kafka-topics.sh --bootstrap-server <broker> --describe --topic <topic>`
2. Fix the `topic` option in the WITH clause (typos, case sensitivity)
3. Confirm bootstrap_servers points at the intended cluster
4. If relying on auto-creation, enable auto.create.topics.enable on the broker or pre-create the topic

// before creating the source
kafka_topics(&bootstrap_servers).contains(&topic)
  .then_some(())
  .ok_or(format!("topic {} missing on cluster", topic))?;

When it happens

Trigger: `fetch_topic_partition` (called by `list_splits`/`list_splits_batch` during source startup or split discovery) fetches metadata for `self.topic` and the response's `topics()` slice is not exactly one element — typically a `None` topic (auto topic creation disabled and topic missing) rather than a single entry.

Common situations: Typo in the topic name in WITH options; topic deleted after source creation; Kafka broker with auto.create.topics.enable=false returning unknown topic; wrong cluster (staging vs prod bootstrap servers); topic name case mismatch.

Understand the failure class

Background: "Not found" and "does not exist" errors: why "Task not found", "No such folder", and "Can't find" fire when a lookup comes back empty — this error's family across 14 libraries.

Related errors

• Iceberg snapshot {snapshot_id} not found
• no fragment connection from source fragment {} to source bac
• sink fragment not found for sink id {}
• secret not found: {0}
• Fail to convert version_hint from utf8 to string: {}

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