fixup! dekaf: Remove unneccesary ensure_topics routine

crates/dekaf/src/api_client.rs

@@ -538,4 +538,59 @@ impl KafkaApiClient {
 
         Ok(version.to_owned())
     }
+
+    #[instrument(skip_all)]
+    pub async fn ensure_topics(&self, topic_names: Vec<messages::TopicName>) -> anyhow::Result<()> {
+        let req = messages::MetadataRequest::default()
+            .with_topics(Some(
+                topic_names
+                    .iter()
+                    .map(|name| {
+                        messages::metadata_request::MetadataRequestTopic::default()
+                            .with_name(Some(name.clone()))
+                    })
+                    .collect(),
+            ))
+            .with_allow_auto_topic_creation(true);
+
+        let coord = self.connect_to_controller().await?;
+        let resp = coord.send_request(req, None).await?;
+        tracing::debug!(metadata=?resp, "Got metadata response");
+
+        if resp
+            .topics
+            .iter()
+            .all(|(name, topic)| topic_names.contains(&name) && topic.error_code == 0)
+        {
+            return Ok(());
+        } else {
+            let mut topics_map = kafka_protocol::indexmap::IndexMap::new();
+            for topic_name in topic_names.into_iter() {
+                topics_map.insert(
+                    topic_name,
+                    messages::create_topics_request::CreatableTopic::default()
+                        .with_replication_factor(2)
+                        .with_num_partitions(-1),
+                );
+            }
+            let create_req = messages::CreateTopicsRequest::default().with_topics(topics_map);
+            let create_resp = coord.send_request(create_req, None).await?;
+            tracing::debug!(create_response=?create_resp, "Got create response");
+
+            for (name, topic) in create_resp.topics {
+                if topic.error_code > 0 {
+                    let err = kafka_protocol::ResponseError::try_from_code(topic.error_code);
+                    tracing::warn!(
+                        topic = name.to_string(),
+                        error = ?err,
+                        message = topic.error_message.map(|m|m.to_string()),
+                        "Failed to create topic"
+                    );
+                    bail!("Failed to create topic");
+                }
+            }
+
+            Ok(())
+        }
+    }
 }

crates/dekaf/src/session.rs

@@ -844,6 +844,17 @@ impl Session {
             .connect_to_group_coordinator(req.group_id.as_str())
             .await?;
 
+        self.app
+            .kafka_client
+            .ensure_topics(
+                mutated_req
+                    .topics
+                    .iter()
+                    .map(|t| t.name.to_owned())
+                    .collect(),
+            )
+            .await?;
+
         let mut resp = client.send_request(mutated_req, Some(header)).await?;
 
         for topic in resp.topics.iter_mut() {
@@ -872,6 +883,13 @@ impl Session {
             .connect_to_group_coordinator(req.group_id.as_str())
             .await?;
 
+        if let Some(ref topics) = mutated_req.topics {
+            self.app
+                .kafka_client
+                .ensure_topics(topics.iter().map(|t| t.name.to_owned()).collect())
+                .await?;
+        }
+
         let mut resp = client.send_request(mutated_req, Some(header)).await?;
 
         for topic in resp.topics.iter_mut() {