From f88167c3d01f2cf55685d900fc4d99252399bb2e Mon Sep 17 00:00:00 2001 From: Guoyu Su Date: Thu, 20 Aug 2026 08:47:05 +0800 Subject: [PATCH] feat(kafka): add consumer group backlog view --- .../java/com/dbx/agent/kafka/KafkaAgent.java | 203 +++++++++ .../com/dbx/agent/kafka/KafkaAgentTest.java | 82 ++++ .../mq/KafkaConsumerGroupsPanel.vue | 387 ++++++++++++++++++ .../src/components/mq/MqAdminConsole.vue | 14 + .../KafkaConsumerGroupsPanel.spec.ts | 178 ++++++++ apps/desktop/src/i18n/locales/en.ts | 24 ++ apps/desktop/src/i18n/locales/es.ts | 24 ++ apps/desktop/src/i18n/locales/it.ts | 24 ++ apps/desktop/src/i18n/locales/ja.ts | 24 ++ apps/desktop/src/i18n/locales/ko.ts | 24 ++ apps/desktop/src/i18n/locales/pt-BR.ts | 24 ++ apps/desktop/src/i18n/locales/zh-CN.ts | 24 ++ apps/desktop/src/i18n/locales/zh-TW.ts | 24 ++ .../src/lib/__tests__/mq/mqListPanels.spec.ts | 18 + apps/desktop/src/lib/backend/api.ts | 1 + apps/desktop/src/lib/backend/mq-http.ts | 5 + apps/desktop/src/lib/backend/mq-tauri.ts | 5 + .../mq/__tests__/mqConsoleDefaults.spec.ts | 11 + apps/desktop/src/lib/mq/mqConsoleDefaults.ts | 7 +- apps/desktop/src/types/mq.ts | 24 ++ crates/dbx-core/src/mq/adapters/kafka.rs | 32 ++ crates/dbx-core/src/mq/port.rs | 7 + crates/dbx-core/src/mq/service.rs | 8 + crates/dbx-core/src/mq/types.rs | 79 ++++ crates/dbx-web/src/main.rs | 1 + crates/dbx-web/src/routes/mq.rs | 12 + src-tauri/src/commands/mq_cmd.rs | 8 + src-tauri/src/lib.rs | 2 + 28 files changed, 1274 insertions(+), 2 deletions(-) create mode 100644 apps/desktop/src/components/mq/KafkaConsumerGroupsPanel.vue create mode 100644 apps/desktop/src/components/mq/__tests__/KafkaConsumerGroupsPanel.spec.ts 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 649fdc83b2..aa981b86aa 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 @@ -146,6 +146,7 @@ public final class KafkaAgent { case "mq_alter_topic_config" -> alterTopicConfig(params); // Consumer groups case "mq_list_consumer_groups" -> listConsumerGroups(params); + case "mq_get_consumer_group_snapshot" -> getConsumerGroupSnapshot(params); case "mq_describe_consumer_group" -> describeConsumerGroup(params); case "mq_delete_consumer_group" -> deleteConsumerGroup(params); case "mq_reset_consumer_group_offsets" -> resetConsumerGroupOffsets(params); @@ -971,6 +972,208 @@ public final class KafkaAgent { return Collections.singletonMap("groups", result); } + private static Object getConsumerGroupSnapshot(JsonObject params) throws Exception { + AdminClient admin = requireAdmin(); + int timeout = requestTimeout(params); + Collection listings = admin.listConsumerGroups( + new ListConsumerGroupsOptions().timeoutMs(timeout)) + .all().get(timeout, TimeUnit.MILLISECONDS); + + List sortedListings = new ArrayList<>(listings); + sortedListings.sort(Comparator.comparing(ConsumerGroupListing::groupId)); + if (sortedListings.isEmpty()) { + return Collections.singletonMap("groups", Collections.emptyList()); + } + + List groupIds = sortedListings.stream() + .map(ConsumerGroupListing::groupId) + .toList(); + Map descriptions = new HashMap<>(); + Map> committedOffsets = new HashMap<>(); + Map committedOffsetsAvailable = new HashMap<>(); + Map> errors = new HashMap<>(); + + DescribeConsumerGroupsResult described = admin.describeConsumerGroups( + groupIds, + new DescribeConsumerGroupsOptions().timeoutMs(timeout) + ); + for (String groupId : groupIds) { + try { + descriptions.put( + groupId, + described.describedGroups().get(groupId).get(timeout, TimeUnit.MILLISECONDS) + ); + } catch (Exception error) { + snapshotErrors(errors, groupId).add("Consumer group details unavailable: " + normalizeErrorMessage(error)); + } + } + + Map offsetSpecs = new LinkedHashMap<>(); + for (String groupId : groupIds) { + offsetSpecs.put(groupId, new ListConsumerGroupOffsetsSpec()); + } + ListConsumerGroupOffsetsResult offsetsResult = admin.listConsumerGroupOffsets( + offsetSpecs, + new ListConsumerGroupOffsetsOptions().timeoutMs(timeout) + ); + for (String groupId : groupIds) { + try { + Map groupOffsets = offsetsResult + .partitionsToOffsetAndMetadata(groupId) + .get(timeout, TimeUnit.MILLISECONDS); + committedOffsets.put(groupId, groupOffsets == null ? Collections.emptyMap() : groupOffsets); + committedOffsetsAvailable.put(groupId, true); + } catch (Exception error) { + committedOffsets.put(groupId, Collections.emptyMap()); + committedOffsetsAvailable.put(groupId, false); + snapshotErrors(errors, groupId).add("Committed offsets unavailable: " + normalizeErrorMessage(error)); + } + } + + Map> assignedPartitions = new HashMap<>(); + Set allPartitions = new LinkedHashSet<>(); + for (String groupId : groupIds) { + Set assigned = new LinkedHashSet<>(); + ConsumerGroupDescription description = descriptions.get(groupId); + if (description != null) { + for (MemberDescription member : description.members()) { + assigned.addAll(member.assignment().topicPartitions()); + } + } + assignedPartitions.put(groupId, assigned); + allPartitions.addAll(assigned); + allPartitions.addAll(committedOffsets.getOrDefault(groupId, Collections.emptyMap()).keySet()); + } + + Map endOffsets = new HashMap<>(); + Map endOffsetErrors = new HashMap<>(); + if (!allPartitions.isEmpty()) { + Map endOffsetSpecs = new LinkedHashMap<>(); + for (TopicPartition topicPartition : allPartitions) { + endOffsetSpecs.put(topicPartition, OffsetSpec.latest()); + } + try { + ListOffsetsResult latestOffsets = admin.listOffsets( + endOffsetSpecs, + new ListOffsetsOptions().timeoutMs(timeout) + ); + for (TopicPartition topicPartition : allPartitions) { + try { + ListOffsetsResult.ListOffsetsResultInfo info = latestOffsets + .partitionResult(topicPartition) + .get(timeout, TimeUnit.MILLISECONDS); + endOffsets.put(topicPartition, info.offset()); + } catch (Exception error) { + endOffsetErrors.put(topicPartition, normalizeErrorMessage(error)); + } + } + } catch (Exception error) { + String message = normalizeErrorMessage(error); + for (TopicPartition topicPartition : allPartitions) { + endOffsetErrors.put(topicPartition, message); + } + } + } + + List> groups = new ArrayList<>(); + for (ConsumerGroupListing listing : sortedListings) { + String groupId = listing.groupId(); + ConsumerGroupDescription description = descriptions.get(groupId); + Set groupPartitions = new LinkedHashSet<>( + assignedPartitions.getOrDefault(groupId, Collections.emptySet()) + ); + groupPartitions.addAll(committedOffsets.getOrDefault(groupId, Collections.emptyMap()).keySet()); + long unavailableEndOffsetCount = groupPartitions.stream() + .filter(endOffsetErrors::containsKey) + .count(); + if (unavailableEndOffsetCount > 0) { + snapshotErrors(errors, groupId).add( + "End offsets unavailable for " + unavailableEndOffsetCount + " partition(s)" + ); + } + groups.add(consumerGroupSnapshotRow( + groupId, + description != null + ? description.state().name() + : listing.state().map(Enum::name).orElse("UNKNOWN"), + listing.isSimpleConsumerGroup(), + description == null ? null : description.members().size(), + assignedPartitions.getOrDefault(groupId, Collections.emptySet()), + committedOffsets.getOrDefault(groupId, Collections.emptyMap()), + committedOffsetsAvailable.getOrDefault(groupId, false), + endOffsets, + errors.getOrDefault(groupId, new LinkedHashSet<>()) + )); + } + + return Collections.singletonMap("groups", groups); + } + + private static LinkedHashSet snapshotErrors( + Map> errors, + String groupId + ) { + return errors.computeIfAbsent(groupId, ignored -> new LinkedHashSet<>()); + } + + static Map consumerGroupSnapshotRow( + String groupId, + String state, + boolean simpleGroup, + Integer memberCount, + Collection assignedPartitions, + Map committedOffsets, + boolean committedOffsetsAvailable, + Map endOffsets, + Collection errors + ) { + Set topicPartitions = new LinkedHashSet<>(assignedPartitions); + topicPartitions.addAll(committedOffsets.keySet()); + List sortedPartitions = new ArrayList<>(topicPartitions); + sortedPartitions.sort(Comparator + .comparing(TopicPartition::topic) + .thenComparingInt(TopicPartition::partition)); + + Set topics = new TreeSet<>(); + List> partitions = new ArrayList<>(); + long totalLag = 0; + boolean lagAvailable = committedOffsetsAvailable && !sortedPartitions.isEmpty(); + for (TopicPartition topicPartition : sortedPartitions) { + topics.add(topicPartition.topic()); + OffsetAndMetadata committed = committedOffsets.get(topicPartition); + Long currentOffset = committed == null ? null : committed.offset(); + Long endOffset = endOffsets.get(topicPartition); + Long lag = currentOffset == null || endOffset == null + ? null + : Math.max(0, endOffset - currentOffset); + if (lag == null) { + lagAvailable = false; + } else { + totalLag += lag; + } + + Map partition = new LinkedHashMap<>(); + partition.put("topic", topicPartition.topic()); + partition.put("partition", topicPartition.partition()); + partition.put("currentOffset", currentOffset); + partition.put("endOffset", endOffset); + partition.put("lag", lag); + partitions.add(partition); + } + + Map group = new LinkedHashMap<>(); + group.put("groupId", groupId); + group.put("state", state); + group.put("simpleGroup", simpleGroup); + group.put("memberCount", memberCount); + group.put("topics", new ArrayList<>(topics)); + group.put("totalLag", lagAvailable ? totalLag : null); + group.put("lagAvailable", lagAvailable); + group.put("partitions", partitions); + group.put("error", errors.isEmpty() ? null : String.join("; ", errors)); + return group; + } + private static Object describeConsumerGroup(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 1d21b3e54a..bce653e234 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 @@ -29,6 +29,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.admin.AlterConfigOp; import org.apache.kafka.clients.admin.Config; @@ -56,6 +57,87 @@ class KafkaAgentTest { @TempDir Path tempDir; + @Test + void consumerGroupSnapshotCalculatesLagAndSortsTopicsAndPartitions() { + TopicPartition alphaOne = new TopicPartition("alpha", 1); + TopicPartition alphaZero = new TopicPartition("alpha", 0); + TopicPartition betaZero = new TopicPartition("beta", 0); + + Map group = KafkaAgent.consumerGroupSnapshotRow( + "orders-service", + "STABLE", + false, + 2, + Arrays.asList(betaZero, alphaOne, alphaZero), + Map.of( + alphaZero, new OffsetAndMetadata(8), + alphaOne, new OffsetAndMetadata(20), + betaZero, new OffsetAndMetadata(5) + ), + true, + Map.of(alphaZero, 10L, alphaOne, 24L, betaZero, 6L), + Collections.emptyList() + ); + + assertEquals(List.of("alpha", "beta"), group.get("topics")); + assertEquals(7L, group.get("totalLag")); + assertEquals(true, group.get("lagAvailable")); + assertEquals(2, group.get("memberCount")); + + @SuppressWarnings("unchecked") + List> partitions = (List>) group.get("partitions"); + assertEquals( + List.of("alpha:0", "alpha:1", "beta:0"), + partitions.stream() + .map(partition -> partition.get("topic") + ":" + partition.get("partition")) + .toList() + ); + } + + @Test + void consumerGroupSnapshotDoesNotReportUnknownLagAsZero() { + TopicPartition assignedOnly = new TopicPartition("orders", 0); + Map group = KafkaAgent.consumerGroupSnapshotRow( + "new-consumer", + "EMPTY", + false, + 0, + Collections.singletonList(assignedOnly), + Collections.emptyMap(), + true, + Collections.singletonMap(assignedOnly, 42L), + Collections.emptyList() + ); + + assertEquals(false, group.get("lagAvailable")); + assertNull(group.get("totalLag")); + @SuppressWarnings("unchecked") + List> partitions = (List>) group.get("partitions"); + assertNull(partitions.get(0).get("currentOffset")); + assertNull(partitions.get(0).get("lag")); + } + + @Test + void consumerGroupSnapshotPreservesPartialFailuresWithoutInventingLag() { + TopicPartition partition = new TopicPartition("orders", 0); + Map group = KafkaAgent.consumerGroupSnapshotRow( + "orders-service", + "UNKNOWN", + false, + null, + Collections.emptyList(), + Collections.singletonMap(partition, new OffsetAndMetadata(9)), + true, + Collections.emptyMap(), + Collections.singletonList("End offsets unavailable for 1 partition(s)") + ); + + assertEquals(false, group.get("lagAvailable")); + assertNull(group.get("totalLag")); + assertNull(group.get("memberCount")); + assertEquals("End offsets unavailable for 1 partition(s)", group.get("error")); + } + @Test void explicitConsumerGroupOffsetsAcceptOneBoundedPartitionOffset() { JsonObject params = JsonParser.parseString(""" diff --git a/apps/desktop/src/components/mq/KafkaConsumerGroupsPanel.vue b/apps/desktop/src/components/mq/KafkaConsumerGroupsPanel.vue new file mode 100644 index 0000000000..b3cc20de86 --- /dev/null +++ b/apps/desktop/src/components/mq/KafkaConsumerGroupsPanel.vue @@ -0,0 +1,387 @@ + + + + + diff --git a/apps/desktop/src/components/mq/MqAdminConsole.vue b/apps/desktop/src/components/mq/MqAdminConsole.vue index e19f9736ee..3e34db9067 100644 --- a/apps/desktop/src/components/mq/MqAdminConsole.vue +++ b/apps/desktop/src/components/mq/MqAdminConsole.vue @@ -27,6 +27,7 @@ import TenantsPanel from "./TenantsPanel.vue"; import NamespacesPanel from "./NamespacesPanel.vue"; import TopicsPanel from "./TopicsPanel.vue"; import SubscriptionsPanel from "./SubscriptionsPanel.vue"; +import KafkaConsumerGroupsPanel from "./KafkaConsumerGroupsPanel.vue"; import MonitoringPanel from "./MonitoringPanel.vue"; import ProducerConsumerPanel from "./ProducerConsumerPanel.vue"; import PoliciesPanel from "./PoliciesPanel.vue"; @@ -174,6 +175,7 @@ function tabLabelKey(tab: MqTab): string { namespaces: "mqAdmin.tabNamespaces", topics: "mqAdmin.tabTopics", subscriptions: "mqAdmin.tabSubscriptions", + consumerGroups: "mqAdmin.tabConsumerGroups", monitoring: "mqAdmin.tabMonitoring", clients: "mqAdmin.tabClients", producers: "mqAdmin.tabProducers", @@ -304,6 +306,17 @@ function handleNavigateTab(payload: { tab: MqTab; topic?: TopicInfo; subscriptio setActiveTab(payload.tab); } +function handleKafkaConsumerGroupTopicSelected(topic: string) { + selectedTopic.value = { + name: topic, + shortName: topic, + partitioned: false, + persistent: true, + }; + selectedSubscriptionName.value = undefined; + setActiveTab("subscriptions"); +} + function handleSubscriptionSelected(subscription: string) { selectedSubscriptionName.value = subscription; activeTab.value = isRocketMqCluster.value ? "producers" : "clients"; @@ -472,6 +485,7 @@ onMounted(async () => { :supports-expire-messages="canExpireMessages" @subscription-selected="handleSubscriptionSelected" /> + diff --git a/apps/desktop/src/components/mq/__tests__/KafkaConsumerGroupsPanel.spec.ts b/apps/desktop/src/components/mq/__tests__/KafkaConsumerGroupsPanel.spec.ts new file mode 100644 index 0000000000..f24a12c7ff --- /dev/null +++ b/apps/desktop/src/components/mq/__tests__/KafkaConsumerGroupsPanel.spec.ts @@ -0,0 +1,178 @@ +// @vitest-environment happy-dom + +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createApp, h, nextTick, ref, type App } from "vue"; +import type { KafkaConsumerGroupSnapshot } from "@/types/mq"; + +const backend = vi.hoisted(() => ({ + mqGetKafkaConsumerGroupSnapshot: vi.fn(), +})); + +vi.mock("vue-i18n", () => ({ + useI18n: () => ({ t: (key: string) => key }), +})); + +vi.mock("@/lib/backend/api", () => backend); + +import KafkaConsumerGroupsPanel from "@/components/mq/KafkaConsumerGroupsPanel.vue"; + +const SNAPSHOT: KafkaConsumerGroupSnapshot = { + groups: [ + { + groupId: "payments-worker", + state: "STABLE", + simpleGroup: false, + memberCount: 1, + topics: ["payments"], + totalLag: 2, + lagAvailable: true, + partitions: [{ topic: "payments", partition: 0, currentOffset: 8, endOffset: 10, lag: 2 }], + }, + { + groupId: "orders-worker", + state: "STABLE", + simpleGroup: false, + memberCount: 2, + topics: ["orders", "orders-archive"], + totalLag: 10, + lagAvailable: true, + partitions: [ + { topic: "orders", partition: 0, currentOffset: 5, endOffset: 12, lag: 7 }, + { topic: "orders", partition: 1, currentOffset: 6, endOffset: 9, lag: 3 }, + { topic: "orders-archive", partition: 0, currentOffset: 4, endOffset: 4, lag: 0 }, + ], + }, + { + groupId: "unknown-worker", + state: "UNKNOWN", + simpleGroup: false, + topics: ["audit"], + lagAvailable: false, + partitions: [{ topic: "audit", partition: 0, currentOffset: 4 }], + error: "End offsets unavailable for 1 partition(s)", + }, + ], +}; + +let app: App | null = null; +let root: HTMLDivElement | null = null; + +async function flushUi() { + await Promise.resolve(); + await Promise.resolve(); + await nextTick(); +} + +async function mountPanel(onNavigateSubscriptions?: (topic: string) => void) { + root = document.createElement("div"); + document.body.appendChild(root); + app = createApp(KafkaConsumerGroupsPanel, { connectionId: "kafka-1", onNavigateSubscriptions }); + app.mount(root); + await flushUi(); + return root; +} + +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +beforeEach(() => { + backend.mqGetKafkaConsumerGroupSnapshot.mockReset().mockResolvedValue(SNAPSHOT); +}); + +afterEach(() => { + app?.unmount(); + app = null; + root?.remove(); + root = null; +}); + +describe("KafkaConsumerGroupsPanel", () => { + it("loads cluster-wide groups without a selected topic and sorts available lag descending", async () => { + const panel = await mountPanel(); + await vi.waitFor(() => expect(backend.mqGetKafkaConsumerGroupSnapshot).toHaveBeenCalledWith("kafka-1")); + + const rows = [...panel.querySelectorAll("[data-group-id]")]; + expect(rows.map((row) => row.dataset.groupId)).toEqual(["orders-worker", "payments-worker", "unknown-worker"]); + expect(panel.querySelector('[data-testid="kafka-consumer-group-count"]')?.textContent?.trim()).toBe("3 / 3"); + expect(panel.querySelector('[data-testid="kafka-consumer-group-detail"]')?.textContent).toContain("orders"); + expect(panel.querySelector('[data-testid="kafka-consumer-group-detail"]')?.textContent).toContain("7"); + }); + + it("emits the exact topic selected from partition details using keyboard-accessible buttons", async () => { + const navigateSubscriptions = vi.fn(); + const panel = await mountPanel(navigateSubscriptions); + await vi.waitFor(() => expect(panel.querySelectorAll("[data-topic]")).toHaveLength(3)); + + const ordersLink = panel.querySelector('[data-topic="orders"]'); + const archiveLink = panel.querySelector('[data-topic="orders-archive"]'); + expect(ordersLink?.tagName).toBe("BUTTON"); + expect(ordersLink?.getAttribute("aria-label")).toBe("mqKafkaConsumerGroups.openTopicSubscriptions"); + + ordersLink?.click(); + archiveLink?.click(); + expect(navigateSubscriptions).toHaveBeenNthCalledWith(1, "orders"); + expect(navigateSubscriptions).toHaveBeenNthCalledWith(2, "orders-archive"); + }); + + it("searches by topic and keeps unavailable lag visibly distinct from zero", async () => { + const panel = await mountPanel(); + await vi.waitFor(() => expect(panel.querySelectorAll("[data-group-id]")).toHaveLength(3)); + + const search = panel.querySelector('input[type="search"]'); + expect(search).toBeTruthy(); + search!.value = "audit"; + search!.dispatchEvent(new Event("input", { bubbles: true })); + await flushUi(); + + const rows = [...panel.querySelectorAll("[data-group-id]")]; + expect(rows.map((row) => row.dataset.groupId)).toEqual(["unknown-worker"]); + expect(rows[0].classList.contains("selected")).toBe(true); + expect(rows[0].querySelector("td:last-child")?.textContent?.trim()).toBe("-"); + expect(rows[0].querySelector(".partial-warning")?.getAttribute("title")).toContain("End offsets unavailable"); + expect(panel.querySelector('[data-testid="kafka-consumer-group-detail"]')?.textContent).toContain("audit"); + }); + + it("shows empty and request-error states", async () => { + backend.mqGetKafkaConsumerGroupSnapshot.mockResolvedValueOnce({ groups: [] }); + const emptyPanel = await mountPanel(); + await vi.waitFor(() => expect(emptyPanel.textContent).toContain("mqKafkaConsumerGroups.empty")); + + app?.unmount(); + root?.remove(); + app = null; + root = null; + backend.mqGetKafkaConsumerGroupSnapshot.mockRejectedValueOnce(new Error("snapshot unavailable")); + const errorPanel = await mountPanel(); + await vi.waitFor(() => expect(errorPanel.textContent).toContain("snapshot unavailable")); + }); + + it("ignores stale refresh responses", async () => { + const first = deferred(); + backend.mqGetKafkaConsumerGroupSnapshot.mockReturnValueOnce(first.promise).mockResolvedValueOnce({ + groups: [{ ...SNAPSHOT.groups[0], groupId: "fresh-worker" }], + }); + const connectionId = ref("kafka-1"); + root = document.createElement("div"); + document.body.appendChild(root); + app = createApp({ + setup: () => () => h(KafkaConsumerGroupsPanel, { connectionId: connectionId.value }), + }); + app.mount(root); + await flushUi(); + + connectionId.value = "kafka-2"; + await flushUi(); + const panel = root; + await vi.waitFor(() => expect(panel.textContent).toContain("fresh-worker")); + + first.resolve(SNAPSHOT); + await flushUi(); + expect(panel.textContent).toContain("fresh-worker"); + expect(panel.textContent).not.toContain("orders-worker"); + }); +}); diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 3cf81e0b84..57fb45e09d 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -7126,6 +7126,7 @@ export default { tabNamespaces: "Namespaces", tabTopics: "Topics", tabSubscriptions: "Subscriptions", + tabConsumerGroups: "Consumer Groups", tabMonitoring: "Monitoring", tabClients: "Clients", tabMessages: "Messages", @@ -7510,6 +7511,29 @@ export default { subscriptionNameRequired: "Subscription name is required", peekMessageKey: "key={key}", }, + mqKafkaConsumerGroups: { + title: "Consumer Groups", + searchPlaceholder: "Search group or topic", + refresh: "Refresh", + refreshing: "Refreshing...", + loading: "Loading consumer groups...", + empty: "No consumer groups", + noMatches: "No matching consumer groups", + groupId: "Group ID", + state: "State", + members: "Members", + topics: "Topics", + totalLag: "Total lag", + partialData: "Some metrics are unavailable", + partitionDetail: "Partition lag details", + noCommittedOffsets: "No committed offsets or assigned partitions", + topic: "Topic", + openTopicSubscriptions: "Open subscriptions for {topic}", + partition: "Partition", + committedOffset: "Committed offset", + logEndOffset: "Log end offset", + lag: "Lag", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka does not support unloading topics", title: "Producers / Consumers", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index aeba7c9b2c..0be6f1b4f3 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -7065,6 +7065,7 @@ export default withEnglishFallback({ tabNamespaces: "Namespaces", tabTopics: "Temas", tabSubscriptions: "Suscripciones", + tabConsumerGroups: "Grupos de consumidores", tabMonitoring: "Monitoreo", tabClients: "Clientes", tabMessages: "Mensajes", @@ -7445,6 +7446,29 @@ export default withEnglishFallback({ broadcasting: "Difusión", }, }, + mqKafkaConsumerGroups: { + title: "Grupos de consumidores", + searchPlaceholder: "Buscar grupo o tema", + refresh: "Actualizar", + refreshing: "Actualizando...", + loading: "Cargando grupos de consumidores...", + empty: "No hay grupos de consumidores", + noMatches: "No hay grupos coincidentes", + groupId: "ID del grupo", + state: "Estado", + members: "Miembros", + topics: "Temas", + totalLag: "Retraso total", + partialData: "Algunas métricas no están disponibles", + partitionDetail: "Detalles del retraso por partición", + noCommittedOffsets: "No hay offsets confirmados ni particiones asignadas", + topic: "Tema", + openTopicSubscriptions: "Abrir suscripciones de {topic}", + partition: "Partición", + committedOffset: "Offset confirmado", + logEndOffset: "Offset final del log", + lag: "Retraso", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka no admite la descarga de temas", title: "Productores / Consumidores", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index bbb6ed34d3..0ad56071a9 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -7065,6 +7065,7 @@ export default withEnglishFallback({ tabNamespaces: "Namespace", tabTopics: "Topic", tabSubscriptions: "Sottoscrizioni", + tabConsumerGroups: "Gruppi di consumatori", tabMonitoring: "Monitoraggio", tabClients: "Clienti", tabMessages: "Messaggi", @@ -7445,6 +7446,29 @@ export default withEnglishFallback({ broadcasting: "Broadcast", }, }, + mqKafkaConsumerGroups: { + title: "Gruppi di consumatori", + searchPlaceholder: "Cerca gruppo o argomento", + refresh: "Aggiorna", + refreshing: "Aggiornamento...", + loading: "Caricamento gruppi di consumatori...", + empty: "Nessun gruppo di consumatori", + noMatches: "Nessun gruppo corrispondente", + groupId: "ID gruppo", + state: "Stato", + members: "Membri", + topics: "Argomenti", + totalLag: "Ritardo totale", + partialData: "Alcune metriche non sono disponibili", + partitionDetail: "Dettagli ritardo partizioni", + noCommittedOffsets: "Nessun offset confermato o partizione assegnata", + topic: "Argomento", + openTopicSubscriptions: "Apri le sottoscrizioni di {topic}", + partition: "Partizione", + committedOffset: "Offset confermato", + logEndOffset: "Offset finale del log", + lag: "Ritardo", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka non supporta lo scaricamento dei topic", title: "Produttori / Consumatori", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index 1664ad8187..634867e3fb 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -7118,6 +7118,7 @@ export default withEnglishFallback({ tabNamespaces: "名前空間", tabTopics: "トピック", tabSubscriptions: "サブスクリプション", + tabConsumerGroups: "コンシューマーグループ", tabMonitoring: "モニタリング", tabClients: "クライアント", tabMessages: "メッセージ", @@ -7498,6 +7499,29 @@ export default withEnglishFallback({ broadcasting: "ブロードキャスト", }, }, + mqKafkaConsumerGroups: { + title: "コンシューマーグループ", + searchPlaceholder: "グループまたはトピックを検索", + refresh: "更新", + refreshing: "更新中...", + loading: "コンシューマーグループを読み込み中...", + empty: "コンシューマーグループがありません", + noMatches: "一致するコンシューマーグループがありません", + groupId: "グループ ID", + state: "状態", + members: "メンバー", + topics: "トピック", + totalLag: "合計ラグ", + partialData: "一部のメトリクスを取得できません", + partitionDetail: "パーティション別ラグ", + noCommittedOffsets: "コミット済みオフセットまたは割り当て済みパーティションがありません", + topic: "トピック", + openTopicSubscriptions: "{topic} のサブスクリプションを開く", + partition: "パーティション", + committedOffset: "コミット済みオフセット", + logEndOffset: "ログ末尾オフセット", + lag: "ラグ", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka はトピックのアンロードに対応していません", title: "プロデューサー / コンシューマー", diff --git a/apps/desktop/src/i18n/locales/ko.ts b/apps/desktop/src/i18n/locales/ko.ts index 395f055907..0b060552d9 100644 --- a/apps/desktop/src/i18n/locales/ko.ts +++ b/apps/desktop/src/i18n/locales/ko.ts @@ -6536,6 +6536,7 @@ export default withEnglishFallback({ tabNamespaces: "네임스페이스", tabTopics: "토픽", tabSubscriptions: "구독", + tabConsumerGroups: "컨슈머 그룹", tabMonitoring: "모니터링", tabClients: "클라이언트", tabMessages: "메시지", @@ -6916,6 +6917,29 @@ export default withEnglishFallback({ subscriptionNameRequired: "구독 이름은 필수입니다", peekMessageKey: "키={key}", }, + mqKafkaConsumerGroups: { + title: "컨슈머 그룹", + searchPlaceholder: "그룹 또는 토픽 검색", + refresh: "새로 고침", + refreshing: "새로 고치는 중...", + loading: "컨슈머 그룹을 불러오는 중...", + empty: "컨슈머 그룹이 없습니다", + noMatches: "일치하는 컨슈머 그룹이 없습니다", + groupId: "그룹 ID", + state: "상태", + members: "멤버", + topics: "토픽", + totalLag: "총 지연", + partialData: "일부 지표를 사용할 수 없습니다", + partitionDetail: "파티션 지연 상세", + noCommittedOffsets: "커밋된 오프셋 또는 할당된 파티션이 없습니다", + topic: "토픽", + openTopicSubscriptions: "{topic} 구독 열기", + partition: "파티션", + committedOffset: "커밋된 오프셋", + logEndOffset: "로그 끝 오프셋", + lag: "지연", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka는 토픽 언로드를 지원하지 않습니다", title: "프로듀서 / 컨슈머", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index 2df3b6a895..84f7b79768 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -7067,6 +7067,7 @@ export default withEnglishFallback({ tabNamespaces: "Namespaces", tabTopics: "Tópicos", tabSubscriptions: "Assinaturas", + tabConsumerGroups: "Grupos de consumidores", tabMonitoring: "Monitoramento", tabClients: "Clientes", tabMessages: "Mensagens", @@ -7447,6 +7448,29 @@ export default withEnglishFallback({ broadcasting: "Broadcast", }, }, + mqKafkaConsumerGroups: { + title: "Grupos de consumidores", + searchPlaceholder: "Pesquisar grupo ou tópico", + refresh: "Atualizar", + refreshing: "Atualizando...", + loading: "Carregando grupos de consumidores...", + empty: "Nenhum grupo de consumidores", + noMatches: "Nenhum grupo correspondente", + groupId: "ID do grupo", + state: "Estado", + members: "Membros", + topics: "Tópicos", + totalLag: "Atraso total", + partialData: "Algumas métricas não estão disponíveis", + partitionDetail: "Detalhes do atraso por partição", + noCommittedOffsets: "Nenhum offset confirmado ou partição atribuída", + topic: "Tópico", + openTopicSubscriptions: "Abrir assinaturas de {topic}", + partition: "Partição", + committedOffset: "Offset confirmado", + logEndOffset: "Offset final do log", + lag: "Atraso", + }, mqClients: { unloadTopicUnsupportedKafka: "O Kafka não suporta o descarregamento de tópicos", title: "Produtores / Consumidores", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 24f5d44122..f801aa0f17 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -7111,6 +7111,7 @@ export default withEnglishFallback({ tabNamespaces: "命名空间", tabTopics: "主题", tabSubscriptions: "订阅", + tabConsumerGroups: "消费组", tabMonitoring: "监控", tabClients: "客户端", tabMessages: "消息", @@ -7495,6 +7496,29 @@ export default withEnglishFallback({ subscriptionNameRequired: "订阅名称不能为空", peekMessageKey: "key={key}", }, + mqKafkaConsumerGroups: { + title: "消费组", + searchPlaceholder: "搜索消费组或 Topic", + refresh: "刷新", + refreshing: "刷新中...", + loading: "正在加载消费组...", + empty: "暂无消费组", + noMatches: "没有匹配的消费组", + groupId: "消费组 ID", + state: "状态", + members: "在线成员", + topics: "Topic", + totalLag: "总积压", + partialData: "部分指标暂不可用", + partitionDetail: "分区积压详情", + noCommittedOffsets: "暂无已提交 Offset 或已分配分区", + topic: "Topic", + openTopicSubscriptions: "打开 {topic} 的订阅", + partition: "分区", + committedOffset: "已提交 Offset", + logEndOffset: "Log End Offset", + lag: "积压", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka 不支持卸载主题", title: "生产者 / 消费者", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index e32767a0fa..00d724fae8 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -7058,6 +7058,7 @@ export default withEnglishFallback({ tabNamespaces: "命名空間", tabTopics: "主題", tabSubscriptions: "訂閱", + tabConsumerGroups: "消費群組", tabMonitoring: "監控", tabClients: "用戶端", tabMessages: "訊息", @@ -7438,6 +7439,29 @@ export default withEnglishFallback({ broadcasting: "廣播", }, }, + mqKafkaConsumerGroups: { + title: "消費群組", + searchPlaceholder: "搜尋消費群組或 Topic", + refresh: "重新整理", + refreshing: "重新整理中...", + loading: "正在載入消費群組...", + empty: "暫無消費群組", + noMatches: "沒有符合的消費群組", + groupId: "消費群組 ID", + state: "狀態", + members: "線上成員", + topics: "Topic", + totalLag: "總積壓", + partialData: "部分指標暫不可用", + partitionDetail: "分區積壓詳情", + noCommittedOffsets: "暫無已提交 Offset 或已分配分區", + topic: "Topic", + openTopicSubscriptions: "開啟 {topic} 的訂閱", + partition: "分區", + committedOffset: "已提交 Offset", + logEndOffset: "Log End Offset", + lag: "積壓", + }, mqClients: { unloadTopicUnsupportedKafka: "Kafka 不支援卸載主題", title: "生產者 / 消費者", diff --git a/apps/desktop/src/lib/__tests__/mq/mqListPanels.spec.ts b/apps/desktop/src/lib/__tests__/mq/mqListPanels.spec.ts index 4536bc6bf3..6e3c9eaab4 100644 --- a/apps/desktop/src/lib/__tests__/mq/mqListPanels.spec.ts +++ b/apps/desktop/src/lib/__tests__/mq/mqListPanels.spec.ts @@ -42,4 +42,22 @@ describe("MQ list panels", () => { expect(consoleSource).toContain(':namespace="selectedNamespace"'); expect(consoleSource).toContain(':topic="selectedTopic"'); }); + + it("keeps Kafka topic subscriptions and cluster-wide consumer groups on separate panels", () => { + const consoleSource = componentSource("MqAdminConsole.vue"); + + expect(consoleSource).toContain("KafkaConsumerGroupsPanel"); + expect(consoleSource).toContain("activeTab === 'subscriptions' && canManageSubscriptions"); + expect(consoleSource).toContain("activeTab === 'consumerGroups' && isKafkaCluster"); + expect(consoleSource).not.toContain("activeTab === 'subscriptions' && canManageSubscriptions && isKafkaCluster"); + }); + + it("routes a Kafka consumer-group partition topic to its topic-scoped subscriptions", () => { + const consoleSource = componentSource("MqAdminConsole.vue"); + + expect(consoleSource).toContain("function handleKafkaConsumerGroupTopicSelected(topic: string)"); + expect(consoleSource).toContain("shortName: topic"); + expect(consoleSource).toContain('setActiveTab("subscriptions")'); + expect(consoleSource).toContain('@navigate-subscriptions="handleKafkaConsumerGroupTopicSelected"'); + }); }); diff --git a/apps/desktop/src/lib/backend/api.ts b/apps/desktop/src/lib/backend/api.ts index 0e8e14bc4a..32a134eed4 100644 --- a/apps/desktop/src/lib/backend/api.ts +++ b/apps/desktop/src/lib/backend/api.ts @@ -639,6 +639,7 @@ export const mqListClientChannels = forward("mqListClientChannels"); export const mqCloseClientConnection = forward("mqCloseClientConnection"); export const mqListSubscriptions = forward("mqListSubscriptions"); export const mqEnrichSubscriptions = forward("mqEnrichSubscriptions"); +export const mqGetKafkaConsumerGroupSnapshot = forward("mqGetKafkaConsumerGroupSnapshot"); export const mqCreateSubscription = forward("mqCreateSubscription"); export const mqDeleteSubscription = forward("mqDeleteSubscription"); export const mqSkipMessages = forward("mqSkipMessages"); diff --git a/apps/desktop/src/lib/backend/mq-http.ts b/apps/desktop/src/lib/backend/mq-http.ts index 7859062197..8c5291ac57 100644 --- a/apps/desktop/src/lib/backend/mq-http.ts +++ b/apps/desktop/src/lib/backend/mq-http.ts @@ -13,6 +13,7 @@ import type { ListTopicsOpts, TopicStats, SubscriptionInfo, + KafkaConsumerGroupSnapshot, ResetPosition, SkipCount, ConsumerInfo, @@ -221,6 +222,10 @@ export async function mqEnrichSubscriptions(connectionId: string, topic: TopicRe return post("/api/mq/subscriptions/enrich", { connectionId, topic }); } +export async function mqGetKafkaConsumerGroupSnapshot(connectionId: string): Promise { + return post("/api/mq/kafka/consumer-groups", { connectionId }); +} + export async function mqCreateSubscription(connectionId: string, topic: TopicRef, sub: string, pos: ResetPosition): Promise { return post("/api/mq/subscriptions/create", { connectionId, topic, sub, pos }); } diff --git a/apps/desktop/src/lib/backend/mq-tauri.ts b/apps/desktop/src/lib/backend/mq-tauri.ts index 7adf51524b..9a545a36a1 100644 --- a/apps/desktop/src/lib/backend/mq-tauri.ts +++ b/apps/desktop/src/lib/backend/mq-tauri.ts @@ -13,6 +13,7 @@ import type { ListTopicsOpts, TopicStats, SubscriptionInfo, + KafkaConsumerGroupSnapshot, ResetPosition, SkipCount, ConsumerInfo, @@ -215,6 +216,10 @@ export async function mqEnrichSubscriptions(connectionId: string, topic: TopicRe return invoke("mq_enrich_subscriptions", { connectionId, topic }); } +export async function mqGetKafkaConsumerGroupSnapshot(connectionId: string): Promise { + return invoke("mq_get_kafka_consumer_group_snapshot", { connectionId }); +} + export async function mqCreateSubscription(connectionId: string, topic: TopicRef, sub: string, pos: ResetPosition): Promise { return invoke("mq_create_subscription", { connectionId, topic, sub, pos }); } diff --git a/apps/desktop/src/lib/mq/__tests__/mqConsoleDefaults.spec.ts b/apps/desktop/src/lib/mq/__tests__/mqConsoleDefaults.spec.ts index 45992f5220..79d678d69e 100644 --- a/apps/desktop/src/lib/mq/__tests__/mqConsoleDefaults.spec.ts +++ b/apps/desktop/src/lib/mq/__tests__/mqConsoleDefaults.spec.ts @@ -85,6 +85,17 @@ describe("mqConsoleDefaults", () => { expect(resolveAvailableMqTabs({ systemKind: "kafka", capabilities: caps })).toContain("messages"); }); + it("keeps Kafka topic subscriptions alongside the cluster-wide consumer groups tab", () => { + const tabs = resolveAvailableMqTabs({ + systemKind: "kafka", + capabilities: defaultMqCapabilitiesForSystemKind("kafka"), + }); + + expect(tabs).toContain("subscriptions"); + expect(tabs).toContain("consumerGroups"); + expect(tabs.indexOf("consumerGroups")).toBe(tabs.indexOf("subscriptions") + 1); + }); + it("exposes the namespaces tab for RabbitMQ vhost management", () => { const caps = defaultMqCapabilitiesForSystemKind("rabbitmq"); expect(resolveAvailableMqTabs({ systemKind: "rabbitmq", capabilities: caps })).toEqual(["namespaces", "topics", "subscriptions", "monitoring", "clients", "messages", "broker", "policies", "permissions"]); diff --git a/apps/desktop/src/lib/mq/mqConsoleDefaults.ts b/apps/desktop/src/lib/mq/mqConsoleDefaults.ts index 80524deb86..19820f326d 100644 --- a/apps/desktop/src/lib/mq/mqConsoleDefaults.ts +++ b/apps/desktop/src/lib/mq/mqConsoleDefaults.ts @@ -158,7 +158,7 @@ export function resolveRabbitMqDefaultVhost(config: ConnectionConfig | undefined return typeof vhost === "string" && vhost.trim() ? vhost.trim() : "/"; } -export type MqTab = "tenants" | "namespaces" | "topics" | "subscriptions" | "monitoring" | "clients" | "producers" | "policies" | "permissions" | "messages" | "raw" | "broker" | "dlq" | "trace"; +export type MqTab = "tenants" | "namespaces" | "topics" | "subscriptions" | "consumerGroups" | "monitoring" | "clients" | "producers" | "policies" | "permissions" | "messages" | "raw" | "broker" | "dlq" | "trace"; export function resolveAvailableMqTabs(options: { systemKind?: MqSystemKind; capabilities: MqCapabilities }): MqTab[] { const { systemKind, capabilities } = options; @@ -174,7 +174,10 @@ export function resolveAvailableMqTabs(options: { systemKind?: MqSystemKind; cap if (capabilities.supportsTenants) tabs.push("tenants"); if (capabilities.supportsNamespaces) tabs.push("namespaces"); tabs.push("topics"); - if (capabilities.supportsSubscriptions) tabs.push("subscriptions"); + if (capabilities.supportsSubscriptions) { + tabs.push("subscriptions"); + if (systemKind === "kafka") tabs.push("consumerGroups"); + } tabs.push("monitoring"); tabs.push("clients"); if (capabilities.supportsSendMessage || (systemKind === "kafka" && capabilities.supportsPeekMessages)) tabs.push("messages"); diff --git a/apps/desktop/src/types/mq.ts b/apps/desktop/src/types/mq.ts index 082c718dd5..7789e6b470 100644 --- a/apps/desktop/src/types/mq.ts +++ b/apps/desktop/src/types/mq.ts @@ -222,6 +222,30 @@ export interface SubscriptionInfo { backlogUnavailable?: boolean; } +export interface KafkaConsumerGroupPartitionLag { + topic: string; + partition: number; + currentOffset?: number; + endOffset?: number; + lag?: number; +} + +export interface KafkaConsumerGroupSummary { + groupId: string; + state: string; + simpleGroup: boolean; + memberCount?: number; + topics: string[]; + totalLag?: number; + lagAvailable: boolean; + partitions: KafkaConsumerGroupPartitionLag[]; + error?: string; +} + +export interface KafkaConsumerGroupSnapshot { + groups: KafkaConsumerGroupSummary[]; +} + export interface RocketMqConsumerGroupConfig { groupName: string; consumeEnable: boolean; diff --git a/crates/dbx-core/src/mq/adapters/kafka.rs b/crates/dbx-core/src/mq/adapters/kafka.rs index e1238465f6..00936e550e 100644 --- a/crates/dbx-core/src/mq/adapters/kafka.rs +++ b/crates/dbx-core/src/mq/adapters/kafka.rs @@ -301,6 +301,10 @@ impl MessageQueueAdmin for KafkaAdmin { Ok(subs) } + async fn get_kafka_consumer_group_snapshot(&self) -> Result { + self.call("mq_get_consumer_group_snapshot", consumer_group_snapshot_params(&self.config)).await + } + async fn create_subscription(&self, _topic: &TopicRef, _sub: &str, _pos: ResetPosition) -> Result<(), String> { Err("Kafka consumer groups are created automatically when consumers join".to_string()) } @@ -657,6 +661,12 @@ fn build_connection_params(cfg: &MqAdminConfig) -> serde_json::Value { }) } +fn consumer_group_snapshot_params(cfg: &MqAdminConfig) -> serde_json::Value { + serde_json::json!({ + "timeout_ms": cfg.request_timeout_ms(), + }) +} + fn peek_messages_params( cfg: &MqAdminConfig, topic: &TopicRef, @@ -864,6 +874,28 @@ mod tests { assert_eq!(params.get("offset").and_then(|value| value.as_i64()), Some(17)); } + #[test] + fn consumer_group_snapshot_params_forward_the_configured_query_timeout() { + let mut cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false); + cfg.query_timeout_secs = 17; + + let params = consumer_group_snapshot_params(&cfg); + + assert_eq!(params.get("timeout_ms").and_then(serde_json::Value::as_u64), Some(17_000)); + assert_eq!(cfg.rpc_timeout(), Some(std::time::Duration::from_secs(17))); + } + + #[test] + fn consumer_group_snapshot_params_keep_a_finite_agent_budget_for_unlimited_rpc_timeout() { + let mut cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false); + cfg.query_timeout_secs = 0; + + let params = consumer_group_snapshot_params(&cfg); + + assert_eq!(params.get("timeout_ms").and_then(serde_json::Value::as_u64), Some(3_600_000)); + assert_eq!(cfg.rpc_timeout(), None); + } + #[test] fn peek_message_params_omit_legacy_optional_fields() { let cfg = kafka_config(serde_json::json!({ "bootstrapServers": "broker:9092" }), MqAuth::None, false); diff --git a/crates/dbx-core/src/mq/port.rs b/crates/dbx-core/src/mq/port.rs index dbc8c7866b..266c0a1524 100644 --- a/crates/dbx-core/src/mq/port.rs +++ b/crates/dbx-core/src/mq/port.rs @@ -156,6 +156,13 @@ pub trait MessageQueueAdmin: Send + Sync { self.list_subscriptions(topic).await } + /// Kafka-only cluster-wide consumer group snapshot. Other MQ systems keep + /// the default unsupported response instead of leaking Kafka concepts into + /// their existing subscription APIs. + async fn get_kafka_consumer_group_snapshot(&self) -> Result { + Err("Kafka consumer group snapshots are not supported by this MQ system".to_string()) + } + async fn create_subscription(&self, topic: &TopicRef, sub: &str, pos: ResetPosition) -> Result<(), String>; async fn delete_subscription(&self, topic: &TopicRef, sub: &str, force: bool) -> Result<(), String>; async fn skip_messages(&self, topic: &TopicRef, sub: &str, count: SkipCount) -> Result<(), String>; diff --git a/crates/dbx-core/src/mq/service.rs b/crates/dbx-core/src/mq/service.rs index f58af5f5a7..6bbefcee6a 100644 --- a/crates/dbx-core/src/mq/service.rs +++ b/crates/dbx-core/src/mq/service.rs @@ -399,6 +399,14 @@ pub async fn mq_enrich_subscriptions_core( adapter.enrich_subscriptions(&topic).await } +pub async fn mq_get_kafka_consumer_group_snapshot_core( + state: &AppState, + conn_id: &str, +) -> Result { + let adapter = get_adapter(state, conn_id).await?; + adapter.get_kafka_consumer_group_snapshot().await +} + pub async fn mq_create_subscription_core( state: &AppState, conn_id: &str, diff --git a/crates/dbx-core/src/mq/types.rs b/crates/dbx-core/src/mq/types.rs index 561197a3b1..eb9400562f 100644 --- a/crates/dbx-core/src/mq/types.rs +++ b/crates/dbx-core/src/mq/types.rs @@ -465,6 +465,50 @@ pub struct SubscriptionInfo { pub backlog_unavailable: Option, } +/// One Kafka topic-partition's committed and log-end offsets for a consumer group. +/// Optional offsets distinguish unavailable data from a healthy zero lag. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct KafkaConsumerGroupPartitionLag { + pub topic: String, + pub partition: i32, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub current_offset: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub end_offset: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub lag: Option, +} + +/// Cluster-wide Kafka consumer group summary with its partition lag details. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct KafkaConsumerGroupSummary { + pub group_id: String, + pub state: String, + #[serde(default)] + pub simple_group: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub member_count: Option, + #[serde(default)] + pub topics: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_lag: Option, + #[serde(default)] + pub lag_available: bool, + #[serde(default)] + pub partitions: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +pub struct KafkaConsumerGroupSnapshot { + #[serde(default)] + pub groups: Vec, +} + #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ConsumerInfo { @@ -521,6 +565,41 @@ mod reset_position_tests { } } +#[cfg(test)] +mod kafka_consumer_group_snapshot_tests { + use super::KafkaConsumerGroupSnapshot; + + #[test] + fn preserves_unavailable_offsets_as_none() { + let snapshot: KafkaConsumerGroupSnapshot = serde_json::from_value(serde_json::json!({ + "groups": [{ + "groupId": "orders-service", + "state": "STABLE", + "simpleGroup": false, + "memberCount": 1, + "topics": ["orders"], + "totalLag": null, + "lagAvailable": false, + "partitions": [{ + "topic": "orders", + "partition": 0, + "currentOffset": 8, + "endOffset": null, + "lag": null + }], + "error": "End offsets unavailable for 1 partition(s)" + }] + })) + .expect("deserialize Kafka consumer group snapshot"); + + assert_eq!(snapshot.groups[0].total_lag, None); + assert!(!snapshot.groups[0].lag_available); + assert_eq!(snapshot.groups[0].partitions[0].current_offset, Some(8)); + assert_eq!(snapshot.groups[0].partitions[0].end_offset, None); + assert_eq!(snapshot.groups[0].partitions[0].lag, None); + } +} + /// How many messages to skip on a subscription. #[derive(Debug, Clone, Copy, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "camelCase", rename_all_fields = "camelCase")] diff --git a/crates/dbx-web/src/main.rs b/crates/dbx-web/src/main.rs index f6d311bd60..5b3197a68a 100644 --- a/crates/dbx-web/src/main.rs +++ b/crates/dbx-web/src/main.rs @@ -181,6 +181,7 @@ fn add_mq_routes(router: Router>) -> Router> { .route("/mq/messages/trace", post(routes::mq::query_message_trace)) .route("/mq/subscriptions/list", post(routes::mq::list_subscriptions)) .route("/mq/subscriptions/enrich", post(routes::mq::enrich_subscriptions)) + .route("/mq/kafka/consumer-groups", post(routes::mq::get_kafka_consumer_group_snapshot)) .route("/mq/subscriptions/create", post(routes::mq::create_subscription)) .route("/mq/subscriptions/delete", post(routes::mq::delete_subscription)) .route("/mq/subscriptions/skip-messages", post(routes::mq::skip_messages)) diff --git a/crates/dbx-web/src/routes/mq.rs b/crates/dbx-web/src/routes/mq.rs index de792d4a1e..60f975e27f 100644 --- a/crates/dbx-web/src/routes/mq.rs +++ b/crates/dbx-web/src/routes/mq.rs @@ -763,6 +763,18 @@ pub async fn enrich_subscriptions( Ok(Json(result)) } +pub async fn get_kafka_consumer_group_snapshot( + State(state): State>, + headers: HeaderMap, + Json(req): Json, +) -> Result, AppError> { + super::mcp_policy::ensure_scope(&state, &headers, &req.connection_id).await?; + let result = dbx_core::mq::service::mq_get_kafka_consumer_group_snapshot_core(&state.app, &req.connection_id) + .await + .map_err(AppError::from)?; + Ok(Json(result)) +} + pub async fn create_subscription( State(state): State>, headers: HeaderMap, diff --git a/src-tauri/src/commands/mq_cmd.rs b/src-tauri/src/commands/mq_cmd.rs index 06d9493d55..01083d5a89 100644 --- a/src-tauri/src/commands/mq_cmd.rs +++ b/src-tauri/src/commands/mq_cmd.rs @@ -276,6 +276,14 @@ pub async fn mq_enrich_subscriptions( dbx_core::mq::service::mq_enrich_subscriptions_core(&state, &connection_id, topic).await } +#[tauri::command] +pub async fn mq_get_kafka_consumer_group_snapshot( + state: State<'_, Arc>, + connection_id: String, +) -> Result { + dbx_core::mq::service::mq_get_kafka_consumer_group_snapshot_core(&state, &connection_id).await +} + #[tauri::command] pub async fn mq_create_subscription( state: State<'_, Arc>, diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 80ed4121f1..9147f4b58e 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2107,6 +2107,8 @@ pub fn run() { #[cfg(feature = "mq-admin")] commands::mq_cmd::mq_enrich_subscriptions, #[cfg(feature = "mq-admin")] + commands::mq_cmd::mq_get_kafka_consumer_group_snapshot, + #[cfg(feature = "mq-admin")] commands::mq_cmd::mq_create_subscription, #[cfg(feature = "mq-admin")] commands::mq_cmd::mq_delete_subscription,