fix(kafka): list topics on legacy API brokers

Closes #5884

agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java

@@ -609,30 +609,52 @@ public final class KafkaAgent {
     private static Object listTopics(JsonObject params) throws Exception {
         AdminClient admin = requireAdmin();
         int timeout = requestTimeout(params);
-        Set<String> names = admin.listTopics(new ListTopicsOptions().timeoutMs(timeout))
-            .names().get(timeout, TimeUnit.MILLISECONDS);
-        if (names.isEmpty()) {
+        Collection<TopicListing> listings = admin.listTopics(new ListTopicsOptions().timeoutMs(timeout))
+            .listings().get(timeout, TimeUnit.MILLISECONDS);
+        return topicListResult(
+            listings,
+            names -> admin.describeTopics(names).allTopicNames().get(timeout, TimeUnit.MILLISECONDS)
+        );
+    }
+
+    static Object topicListResult(Collection<TopicListing> listings, TopicDescriptionLoader descriptionLoader) throws Exception {
+        if (listings.isEmpty()) {
             return Collections.singletonMap("topics", Collections.emptyList());
         }
 
-        Map<String, TopicDescription> descriptions = admin.describeTopics(names)
-            .allTopicNames().get(timeout, TimeUnit.MILLISECONDS);
-
         List<Map<String, Object>> topics = new ArrayList<>();
-        for (Map.Entry<String, TopicDescription> entry : descriptions.entrySet()) {
-            TopicDescription desc = entry.getValue();
-            Map<String, Object> topic = new LinkedHashMap<>();
-            topic.put("name", desc.name());
-            topic.put("partitions", desc.partitions().size());
-            topic.put("replicationFactor", desc.partitions().isEmpty() ? 0
-                : desc.partitions().get(0).replicas().size());
-            topic.put("internal", desc.isInternal());
-            topics.add(topic);
+        try {
+            Set<String> names = listings.stream().map(TopicListing::name).collect(Collectors.toCollection(LinkedHashSet::new));
+            Map<String, TopicDescription> descriptions = descriptionLoader.load(names);
+            for (TopicDescription desc : descriptions.values()) {
+                Map<String, Object> topic = new LinkedHashMap<>();
+                topic.put("name", desc.name());
+                topic.put("partitions", desc.partitions().size());
+                topic.put("replicationFactor", desc.partitions().isEmpty() ? 0
+                    : desc.partitions().get(0).replicas().size());
+                topic.put("internal", desc.isInternal());
+                topics.add(topic);
+            }
+        } catch (Exception error) {
+            if (!isUnsupportedVersionError(error)) {
+                throw error;
+            }
+            for (TopicListing listing : listings) {
+                Map<String, Object> topic = new LinkedHashMap<>();
+                topic.put("name", listing.name());
+                topic.put("internal", listing.isInternal());
+                topics.add(topic);
+            }
         }
         topics.sort(Comparator.comparing(m -> (String) m.get("name")));
         return Collections.singletonMap("topics", topics);
     }
 
+    @FunctionalInterface
+    interface TopicDescriptionLoader {
+        Map<String, TopicDescription> load(Set<String> names) throws Exception;
+    }
+
     private static Object createTopic(JsonObject params) throws Exception {
         AdminClient admin = requireAdmin();
         int timeout = requestTimeout(params);

agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java

@@ -23,6 +23,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -32,8 +33,13 @@ import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.RaftVoterEndpoint;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.clients.admin.TopicListing;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooDefs;
@@ -48,6 +54,68 @@ class KafkaAgentTest {
     @TempDir
     Path tempDir;
 
+    @Test
+    void topicListingFallsBackWhenDescriptionsAreUnsupported() throws Exception {
+        Object result = KafkaAgent.topicListResult(
+            Arrays.asList(
+                new TopicListing("orders", Uuid.randomUuid(), false),
+                new TopicListing("__consumer_offsets", Uuid.randomUuid(), true)
+            ),
+            names -> {
+                throw new ExecutionException(new UnsupportedVersionException("The version of API is not supported."));
+            }
+        );
+
+        @SuppressWarnings("unchecked")
+        List<Map<String, Object>> topics = (List<Map<String, Object>>) ((Map<String, Object>) result).get("topics");
+        assertEquals(Arrays.asList("__consumer_offsets", "orders"), topics.stream().map(topic -> topic.get("name")).toList());
+        assertEquals(true, topics.get(0).get("internal"));
+        assertFalse(topics.get(0).containsKey("partitions"));
+        assertFalse(topics.get(1).containsKey("partitions"));
+    }
+
+    @Test
+    void topicListingPreservesNonVersionDescriptionErrors() {
+        IllegalStateException failure = new IllegalStateException("metadata authorization failed");
+
+        IllegalStateException thrown = assertThrows(IllegalStateException.class, () -> KafkaAgent.topicListResult(
+            Collections.singletonList(new TopicListing("orders", Uuid.randomUuid(), false)),
+            names -> {
+                throw failure;
+            }
+        ));
+
+        assertEquals(failure, thrown);
+    }
+
+    @Test
+    void topicListingKeepsDescriptionMetadataWhenSupported() throws Exception {
+        Node leader = new Node(1, "broker-1", 9092);
+        Node replica = new Node(2, "broker-2", 9092);
+        TopicDescription description = new TopicDescription(
+            "orders",
+            false,
+            Collections.singletonList(new TopicPartitionInfo(
+                0,
+                leader,
+                Arrays.asList(leader, replica),
+                Collections.singletonList(leader)
+            ))
+        );
+
+        Object result = KafkaAgent.topicListResult(
+            Collections.singletonList(new TopicListing("orders", Uuid.randomUuid(), false)),
+            names -> Collections.singletonMap("orders", description)
+        );
+
+        @SuppressWarnings("unchecked")
+        List<Map<String, Object>> topics = (List<Map<String, Object>>) ((Map<String, Object>) result).get("topics");
+        assertEquals(1, topics.size());
+        assertEquals(1, topics.get(0).get("partitions"));
+        assertEquals(2, topics.get(0).get("replicationFactor"));
+        assertEquals(false, topics.get(0).get("internal"));
+    }
+
     @Test
     void resolvesBootstrapServersFromKafka11ZooKeeperRegistrationWithChroot() throws Exception {
         Path snapshots = Files.createDirectory(tempDir.resolve("snapshots"));