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 97d7ad338e..749f52bf61 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 @@ -704,8 +704,22 @@ public final class KafkaAgent { int timeout = requestTimeout(params); String name = stringOrEmpty(params, "name"); + try { + return modernTopicStats(admin, name, timeout); + } catch (Exception error) { + if (!hasUnsupportedVersionException(error)) { + throw error; + } + return legacyTopicStats(name, timeout); + } + } + + private static Object modernTopicStats(AdminClient admin, String name, int timeout) throws Exception { TopicDescription desc = admin.describeTopics(Collections.singletonList(name)) .allTopicNames().get(timeout, TimeUnit.MILLISECONDS).get(name); + if (desc == null) { + throw new UnknownTopicOrPartitionException("Kafka topic does not exist: " + name); + } // Collect offsets for size estimation Map endOffsets = new LinkedHashMap<>(); @@ -748,14 +762,101 @@ public final class KafkaAgent { return result; } + private static Object legacyTopicStats(String name, int timeout) { + Properties props = topicStatsConsumerProperties(activeConnection); + Duration requestTimeout = Duration.ofMillis(timeout); + try (KafkaConsumer consumer = new KafkaConsumer<>(props)) { + Map> topics = consumer.listTopics(requestTimeout); + requireExistingTopic(topics.keySet(), name); + List partitions = topics.get(name); + List topicPartitions = partitions.stream() + .map(partition -> new TopicPartition(name, partition.partition())) + .collect(Collectors.toList()); + Map beginningOffsets = consumer.beginningOffsets(topicPartitions, requestTimeout); + Map endOffsets = consumer.endOffsets(topicPartitions, requestTimeout); + return legacyTopicStatsResult(name, partitions, beginningOffsets, endOffsets); + } + } + + static void requireExistingTopic(Collection topicNames, String name) { + if (!topicNames.contains(name)) { + throw new UnknownTopicOrPartitionException("Kafka topic does not exist: " + name); + } + } + + static Properties topicStatsConsumerProperties(JsonObject conn) { + if (conn == null) { + throw new IllegalStateException("Kafka Agent is not connected"); + } + Properties props = peekConsumerProperties(conn, 1); + // The fallback discovers the topic through all-topics metadata, which cannot create + // a topic, before issuing requests for its concrete partitions. + props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "true"); + return props; + } + + static Object legacyTopicStatsResult( + String name, + List partitions, + Map beginningOffsets, + Map endOffsets + ) { + long totalMessages = 0; + List> partitionStats = new ArrayList<>(); + for (PartitionInfo partition : partitions) { + TopicPartition topicPartition = new TopicPartition(name, partition.partition()); + long begin = beginningOffsets.get(topicPartition); + long end = endOffsets.get(topicPartition); + long count = end - begin; + totalMessages += count; + + Map stats = new LinkedHashMap<>(); + stats.put("partition", partition.partition()); + stats.put("leader", partition.leader() != null ? partition.leader().id() : -1); + stats.put("replicas", Arrays.stream(partition.replicas()).map(Node::id).collect(Collectors.toList())); + stats.put("isr", Arrays.stream(partition.inSyncReplicas()).map(Node::id).collect(Collectors.toList())); + stats.put("beginOffset", begin); + stats.put("endOffset", end); + stats.put("messageCount", count); + partitionStats.add(stats); + } + + Map result = new LinkedHashMap<>(); + result.put("name", name); + result.put("partitions", partitions.size()); + result.put("replicationFactor", partitions.isEmpty() ? 0 : partitions.get(0).replicas().length); + result.put("totalMessages", totalMessages); + result.put("partitionStats", partitionStats); + return result; + } + private static Object getTopicConfig(JsonObject params) throws Exception { AdminClient admin = requireAdmin(); int timeout = requestTimeout(params); String name = stringOrEmpty(params, "name"); ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, name); - Config config = admin.describeConfigs(Collections.singletonList(resource)) - .all().get(timeout, TimeUnit.MILLISECONDS).get(resource); + return topicConfigResult(() -> admin.describeConfigs(Collections.singletonList(resource)) + .all().get(timeout, TimeUnit.MILLISECONDS).get(resource)); + } + + static Object topicConfigResult(TopicConfigLoader configLoader) throws Exception { + Config config; + try { + config = configLoader.load(); + } catch (Exception error) { + if (!hasUnsupportedVersionException(error)) { + throw error; + } + Map unsupported = new LinkedHashMap<>(); + unsupported.put("configs", Collections.emptyMap()); + unsupported.put("configSupported", false); + unsupported.put( + "unsupportedReason", + "Topic configuration is unavailable because this Kafka broker does not support DescribeConfigs." + ); + return unsupported; + } Map configs = new LinkedHashMap<>(); for (ConfigEntry entry : config.entries()) { @@ -770,6 +871,11 @@ public final class KafkaAgent { return Collections.singletonMap("configs", configs); } + @FunctionalInterface + interface TopicConfigLoader { + Config load() throws Exception; + } + private static Object alterTopicConfig(JsonObject params) throws Exception { AdminClient admin = requireAdmin(); int timeout = requestTimeout(params); @@ -2098,6 +2204,10 @@ public final class KafkaAgent { return false; } + static boolean hasUnsupportedVersionException(Throwable error) { + return causeChain(error).stream().anyMatch(UnsupportedVersionException.class::isInstance); + } + private static Throwable rootCause(Throwable error) { Throwable current = null; for (Throwable cause : causeChain(error)) { 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 6933bb0a84..8af5abbd7a 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 @@ -36,6 +36,7 @@ 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.PartitionInfo; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.TopicPartitionInfo; import org.apache.kafka.common.Uuid; @@ -150,6 +151,114 @@ class KafkaAgentTest { assertEquals(false, topics.get(0).get("internal")); } + @Test + void legacyStatsFallbackRequiresAnUnsupportedVersionException() { + assertTrue(KafkaAgent.hasUnsupportedVersionException(new ExecutionException( + new UnsupportedVersionException( + "MetadataRequest versions older than 4 don't support the allowAutoTopicCreation field" + ) + ))); + assertFalse(KafkaAgent.hasUnsupportedVersionException( + new IllegalStateException("broker reports unsupported version text") + )); + } + + @Test + void legacyTopicStatsRequireAnExistingTopicFromAllTopicMetadata() { + var error = assertThrows( + org.apache.kafka.common.errors.UnknownTopicOrPartitionException.class, + () -> KafkaAgent.requireExistingTopic(Collections.singleton("payments"), "orders") + ); + + assertTrue(error.getMessage().contains("orders")); + } + + @Test + void legacyTopicStatsConsumerUsesCompatibleMetadataOnlyForConnectedSessions() { + assertThrows(IllegalStateException.class, () -> KafkaAgent.topicStatsConsumerProperties(null)); + + JsonObject connection = new JsonObject(); + connection.addProperty("bootstrap_servers", "legacy-broker:9092"); + Properties properties = KafkaAgent.topicStatsConsumerProperties(connection); + + assertEquals("true", properties.getProperty("allow.auto.create.topics")); + assertEquals("false", properties.getProperty("enable.auto.commit")); + } + + @Test + void legacyTopicStatsPreserveTheExistingStatsShape() { + Node leader = new Node(1, "broker-1", 9092); + Node replica = new Node(2, "broker-2", 9092); + PartitionInfo partition = new PartitionInfo( + "orders", + 0, + leader, + new Node[] { leader, replica }, + new Node[] { leader } + ); + TopicPartition topicPartition = new TopicPartition("orders", 0); + + @SuppressWarnings("unchecked") + Map result = (Map) KafkaAgent.legacyTopicStatsResult( + "orders", + Collections.singletonList(partition), + Collections.singletonMap(topicPartition, 4L), + Collections.singletonMap(topicPartition, 10L) + ); + + assertEquals("orders", result.get("name")); + assertEquals(1, result.get("partitions")); + assertEquals(2, result.get("replicationFactor")); + assertEquals(6L, result.get("totalMessages")); + @SuppressWarnings("unchecked") + List> partitionStats = (List>) result.get("partitionStats"); + assertEquals(List.of(1, 2), partitionStats.get(0).get("replicas")); + assertEquals(Collections.singletonList(1), partitionStats.get(0).get("isr")); + assertEquals(4L, partitionStats.get(0).get("beginOffset")); + assertEquals(10L, partitionStats.get(0).get("endOffset")); + } + + @Test + void topicConfigReturnsAnExplicitUnsupportedMarkerForLegacyBrokers() throws Exception { + @SuppressWarnings("unchecked") + Map result = (Map) KafkaAgent.topicConfigResult(() -> { + throw new ExecutionException(new UnsupportedVersionException( + "The node does not support DESCRIBE_CONFIGS" + )); + }); + + assertEquals(Collections.emptyMap(), result.get("configs")); + assertEquals(false, result.get("configSupported")); + assertTrue(((String) result.get("unsupportedReason")).contains("DescribeConfigs")); + } + + @Test + void topicConfigPreservesModernConfigEntries() throws Exception { + Config config = new Config(Collections.singletonList(new ConfigEntry("retention.ms", "60000"))); + + @SuppressWarnings("unchecked") + Map result = (Map) KafkaAgent.topicConfigResult(() -> config); + @SuppressWarnings("unchecked") + Map> configs = (Map>) result.get("configs"); + + assertEquals("60000", configs.get("retention.ms").get("value")); + assertFalse(result.containsKey("configSupported")); + } + + @Test + void topicConfigPreservesNonVersionFailures() { + IllegalStateException failure = new IllegalStateException("broker reports unsupported version text"); + + IllegalStateException thrown = assertThrows( + IllegalStateException.class, + () -> KafkaAgent.topicConfigResult(() -> { + throw failure; + }) + ); + + assertEquals(failure, thrown); + } + @Test void resolvesBootstrapServersFromKafka11ZooKeeperRegistrationWithChroot() throws Exception { Path snapshots = Files.createDirectory(tempDir.resolve("snapshots")); diff --git a/apps/desktop/src/components/mq/PoliciesPanel.vue b/apps/desktop/src/components/mq/PoliciesPanel.vue index d8483523d3..faca099288 100644 --- a/apps/desktop/src/components/mq/PoliciesPanel.vue +++ b/apps/desktop/src/components/mq/PoliciesPanel.vue @@ -4,7 +4,7 @@ import { computed, ref, watch } from "vue"; import { useI18n } from "vue-i18n"; import type { BacklogQuota, DispatchRate, PolicyScope, PublishRate, RetentionPolicy, SubscribeRate, TopicInfo } from "@/types/mq"; import { mqGetEffectivePolicies, mqSetBacklogQuota, mqSetDispatchRate, mqSetPublishRate, mqSetRetention, mqSetSubscribeRate } from "@/lib/backend/api"; -import { defaultMqPolicyForms, policyFormsFromEffectivePolicies } from "@/lib/mq/mqPolicyForms"; +import { defaultMqPolicyForms, policyAccessFromEffectivePolicies, policyFormsFromEffectivePolicies } from "@/lib/mq/mqPolicyForms"; import { useMqMutationGuard } from "@/composables/useMqMutationGuard"; interface Props { @@ -29,6 +29,7 @@ const error = ref(); const notice = ref(); const readOnlyMessage = computed(() => t("mqPolicies.readOnly")); const defaultForms = defaultMqPolicyForms(); +const policyAccess = ref(policyAccessFromEffectivePolicies(undefined)); const publishForm = ref({ ...defaultForms.publishForm }); @@ -84,6 +85,7 @@ async function guardWritable(operation: string): Promise { async function loadPolicies() { const current = scope.value; policies.value = undefined; + policyAccess.value = policyAccessFromEffectivePolicies(undefined); notice.value = undefined; error.value = undefined; if (!current) return; @@ -92,6 +94,7 @@ async function loadPolicies() { try { const loaded = await mqGetEffectivePolicies(props.connectionId, current); policies.value = loaded; + policyAccess.value = policyAccessFromEffectivePolicies(loaded); const hydrated = policyFormsFromEffectivePolicies(loaded, defaultMqPolicyForms()); applyPolicyForms(hydrated); } catch (e: unknown) { @@ -189,7 +192,8 @@ watch(
{{ error }}
{{ notice }}
-
+
{{ policyAccess.unsupportedReason }}
+

{{ t("mqPolicies.publishRate") }}