diff --git a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java b/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java index 36b3bf1a01..ff29fbe8e3 100644 --- a/agents/drivers/kafka/src/main/java/com/dbx/agent/kafka/KafkaAgent.java +++ b/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 names = admin.listTopics(new ListTopicsOptions().timeoutMs(timeout)) - .names().get(timeout, TimeUnit.MILLISECONDS); - if (names.isEmpty()) { + Collection 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 listings, TopicDescriptionLoader descriptionLoader) throws Exception { + if (listings.isEmpty()) { return Collections.singletonMap("topics", Collections.emptyList()); } - Map descriptions = admin.describeTopics(names) - .allTopicNames().get(timeout, TimeUnit.MILLISECONDS); - List> topics = new ArrayList<>(); - for (Map.Entry entry : descriptions.entrySet()) { - TopicDescription desc = entry.getValue(); - Map 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 names = listings.stream().map(TopicListing::name).collect(Collectors.toCollection(LinkedHashSet::new)); + Map descriptions = descriptionLoader.load(names); + for (TopicDescription desc : descriptions.values()) { + Map 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 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 load(Set names) throws Exception; + } + private static Object createTopic(JsonObject params) throws Exception { AdminClient admin = requireAdmin(); int timeout = requestTimeout(params); diff --git a/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java b/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java index 654f61f9a2..360d079a32 100644 --- a/agents/drivers/kafka/src/test/java/com/dbx/agent/kafka/KafkaAgentTest.java +++ b/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> topics = (List>) ((Map) 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> topics = (List>) ((Map) 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"));