From ae87353a28830d277e9d293f9ec6b4dcaf564476 Mon Sep 17 00:00:00 2001 From: onenewcode Date: Tue, 4 Aug 2026 13:13:49 +0800 Subject: [PATCH] feat(kafka): redesign message browsing --- .../java/com/dbx/agent/kafka/KafkaAgent.java | 533 ++++++++++++++++-- .../com/dbx/agent/kafka/KafkaAgentTest.java | 445 ++++++++++++++- .../src/components/mq/MessageBrowser.vue | 462 +++++++++++++++ .../src/components/mq/MessageQueryPanel.vue | 3 +- .../src/components/mq/MonitoringPanel.vue | 197 +------ .../src/components/mq/SendMessagePanel.vue | 270 +-------- .../src/components/mq/SubscriptionsPanel.vue | 44 +- .../mq/__tests__/MessageBrowser.spec.ts | 458 +++++++++++++++ .../mq/__tests__/MonitoringPanel.spec.ts | 181 ++++++ .../mq/__tests__/SendMessagePanel.spec.ts | 126 ++++- .../mq/__tests__/SubscriptionsPanel.spec.ts | 178 ++++++ .../src/components/mq/__tests__/selectStub.ts | 36 ++ apps/desktop/src/i18n/locales/en.ts | 18 +- apps/desktop/src/i18n/locales/es.ts | 18 +- apps/desktop/src/i18n/locales/it.ts | 18 +- apps/desktop/src/i18n/locales/ja.ts | 18 +- apps/desktop/src/i18n/locales/ko.ts | 18 +- apps/desktop/src/i18n/locales/pt-BR.ts | 18 +- apps/desktop/src/i18n/locales/zh-CN.ts | 18 +- apps/desktop/src/i18n/locales/zh-TW.ts | 18 +- apps/desktop/src/lib/backend/mq-http.ts | 4 +- apps/desktop/src/lib/backend/mq-tauri.ts | 4 +- apps/desktop/src/types/mq.ts | 8 + crates/dbx-core/src/mq/adapters/kafka.rs | 212 +++++-- crates/dbx-core/src/mq/adapters/pulsar.rs | 6 +- crates/dbx-core/src/mq/adapters/rabbitmq.rs | 6 +- crates/dbx-core/src/mq/adapters/rocketmq.rs | 6 +- crates/dbx-core/src/mq/port.rs | 2 +- crates/dbx-core/src/mq/service.rs | 4 +- crates/dbx-core/src/mq/types.rs | 67 ++- crates/dbx-web/src/routes/mq.rs | 2 +- src-tauri/src/commands/mq_cmd.rs | 2 +- 32 files changed, 2776 insertions(+), 624 deletions(-) create mode 100644 apps/desktop/src/components/mq/MessageBrowser.vue create mode 100644 apps/desktop/src/components/mq/__tests__/MessageBrowser.spec.ts create mode 100644 apps/desktop/src/components/mq/__tests__/MonitoringPanel.spec.ts create mode 100644 apps/desktop/src/components/mq/__tests__/SubscriptionsPanel.spec.ts create mode 100644 apps/desktop/src/components/mq/__tests__/selectStub.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 d546bdbba1..07f64dbec1 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 @@ -41,6 +41,8 @@ public final class KafkaAgent { private static final PrintStream JSON_RPC_OUT = System.out; private static final Gson GSON = new GsonBuilder().serializeNulls().create(); private static final int DEFAULT_REQUEST_TIMEOUT_MS = 30_000; + private static final int MAX_PEEK_MESSAGE_COUNT = 100; + private static final int MAX_PEEK_SCAN_RECORDS = 1_000; private static final int DEFAULT_SESSION_TIMEOUT_MS = 30_000; private static final int DEFAULT_ZOOKEEPER_CONNECTION_TIMEOUT_MS = 10_000; private static final String ZOOKEEPER_PROPERTY_PREFIX = "zookeeper."; @@ -1025,31 +1027,43 @@ public final class KafkaAgent { String topic = stringOrEmpty(params, "topic"); Integer partition = integerOrNull(params, "partition"); Long offset = longOrNull(params, "offset"); - int count = Math.max(1, intOrDefault(params, "count", 10)); + int count = validatedPeekCount(intOrDefault(params, "count", 10)); + PeekStartPosition startPosition = peekStartPosition(params); + boolean explicitStartPosition = stringOrNull(params, "startPosition") != null; + validatePeekRequest(startPosition, explicitStartPosition, partition, offset); + boolean legacyOffsetRequest = !explicitStartPosition && offset != null; JsonObject conn = activeConnection; if (conn == null) { throw new IllegalStateException("Kafka Agent is not connected"); } Properties props = peekConsumerProperties(conn, count); + Duration requestTimeout = Duration.ofMillis(peekRequestTimeoutMs(conn, props)); try (KafkaConsumer consumer = new KafkaConsumer<>(props)) { - List candidatePartitions = resolvePeekPartitions(consumer, topic, partition); + List candidatePartitions = resolvePeekPartitions( + consumer, topic, partition, requestTimeout + ); if (candidatePartitions.isEmpty()) { - return Collections.singletonMap("messages", Collections.emptyList()); + return peekMessagesResult(Collections.emptyList(), false); } Map beginningOffsets = - consumer.beginningOffsets(candidatePartitions, Duration.ofSeconds(5)); + consumer.beginningOffsets(candidatePartitions, requestTimeout); Map endOffsets = - consumer.endOffsets(candidatePartitions, Duration.ofSeconds(5)); + consumer.endOffsets(candidatePartitions, requestTimeout); List readablePartitions = new ArrayList<>(); Map seekOffsets = new LinkedHashMap<>(); for (TopicPartition tp : candidatePartitions) { long beginningOffset = beginningOffsets.getOrDefault(tp, 0L); long endOffset = endOffsets.getOrDefault(tp, beginningOffset); - long requestedOffset = offset != null ? offset : beginningOffset; + Long requestedOffset = requestedPeekOffset( + startPosition, offset, legacyOffsetRequest, beginningOffset, endOffset + ); + if (requestedOffset == null) { + continue; + } Long seekOffset = normalizePeekOffset(requestedOffset, beginningOffset, endOffset); if (seekOffset == null) { continue; @@ -1058,7 +1072,23 @@ public final class KafkaAgent { seekOffsets.put(tp, seekOffset); } if (readablePartitions.isEmpty()) { - return Collections.singletonMap("messages", Collections.emptyList()); + return peekMessagesResult(Collections.emptyList(), false); + } + + int messagesPerPartition = peekMessagesPerPartition(count, readablePartitions.size()); + int scanLimit = peekScanLimit(count, readablePartitions.size()); + Map snapshotEndOffsets = new LinkedHashMap<>(); + if (startPosition == PeekStartPosition.LATEST) { + for (TopicPartition tp : readablePartitions) { + long beginningOffset = beginningOffsets.getOrDefault(tp, 0L); + long endOffset = endOffsets.getOrDefault(tp, beginningOffset); + seekOffsets.put(tp, recentPeekStartOffset( + beginningOffset, endOffset, messagesPerPartition + )); + } + } + for (TopicPartition tp : readablePartitions) { + snapshotEndOffsets.put(tp, endOffsets.getOrDefault(tp, 0L)); } consumer.assign(readablePartitions); @@ -1066,27 +1096,89 @@ public final class KafkaAgent { consumer.seek(entry.getKey(), entry.getValue()); } - List> messages = collectPeekedMessages( - timeout -> consumer.poll(timeout), - () -> { - Map positions = new LinkedHashMap<>(); - for (TopicPartition tp : readablePartitions) { - positions.put(tp, consumer.position(tp)); - } - return allPeekPartitionsCaughtUp(readablePartitions, positions, endOffsets); - }, - count, - System.nanoTime() + Duration.ofSeconds(5).toNanos(), - Duration.ofMillis(500) - ); - sortPeekedMessages(messages); + long deadlineNs = System.nanoTime() + requestTimeout.toNanos(); + Duration pollTimeout = Duration.ofMillis(Math.min(500L, requestTimeout.toMillis())); + PeekCollectionState collection; + if (startPosition == PeekStartPosition.LATEST) { + collection = collectLatestPeekedMessages( + consumer, + readablePartitions, + beginningOffsets, + seekOffsets, + snapshotEndOffsets, + messagesPerPartition, + scanLimit, + deadlineNs, + pollTimeout + ); + } else { + PeekCollectionCompletionChecker snapshotComplete = state -> allPeekPartitionsComplete( + readablePartitions, + state.remainingByPartition, + currentPeekPositions(consumer, readablePartitions), + snapshotEndOffsets + ); + collection = new PeekCollectionState(readablePartitions, messagesPerPartition); + collection.incomplete = !collectPeekedMessages( + timeout -> consumer.poll(timeout), + snapshotComplete, + record -> recordIsBeforeEndOffset(record, snapshotEndOffsets), + collection, + scanLimit, + deadlineNs, + pollTimeout + ); + } + List> messages = collection.messages; + sortPeekedMessages(messages, startPosition); if (messages.size() > count) { messages = new ArrayList<>(messages.subList(0, count)); } - return Collections.singletonMap("messages", messages); + return peekMessagesResult(messages, collection.incomplete); } } + static Map peekMessagesResult(List> messages, boolean incomplete) { + Map result = new LinkedHashMap<>(); + result.put("messages", messages); + result.put("incomplete", incomplete); + return result; + } + + static int validatedPeekCount(int count) { + if (count < 1 || count > MAX_PEEK_MESSAGE_COUNT) { + throw new IllegalArgumentException( + "Peek message count must be between 1 and " + MAX_PEEK_MESSAGE_COUNT + ); + } + return count; + } + + static int peekRequestTimeoutMs(JsonObject conn, Properties props) { + Integer connectionTimeout = integerOrNull(conn, "request_timeout_ms"); + if (connectionTimeout != null) { + return positiveTimeoutMs("request_timeout_ms", connectionTimeout); + } + String configuredTimeout = props.getProperty( + ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, + String.valueOf(DEFAULT_REQUEST_TIMEOUT_MS) + ); + try { + return positiveTimeoutMs(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, Integer.parseInt(configuredTimeout)); + } catch (NumberFormatException error) { + throw new IllegalArgumentException( + ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG + " must be a positive integer", error + ); + } + } + + private static int positiveTimeoutMs(String name, int timeoutMs) { + if (timeoutMs <= 0) { + throw new IllegalArgumentException(name + " must be a positive integer"); + } + return timeoutMs; + } + static Properties peekConsumerProperties(JsonObject conn, int count) { Properties props = new Properties(); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers(conn)); @@ -1103,19 +1195,45 @@ public final class KafkaAgent { } /** - * Poll until {@code count} messages are collected, every assigned partition has reached its - * end offset, or {@code deadlineNs} expires. Empty polls retry until caught-up or deadline — - * they must not abort early (broker / network / first-fetch latency can exceed one poll). + * Reads until each partition supplies its share of the page or reaches the snapshot boundary. + * This counts retained records, rather than treating an offset range as a record count. */ static List> collectPeekedMessages( PeekRecordPoller poller, PeekCaughtUpChecker caughtUpChecker, - int count, + PeekRecordFilter recordFilter, + List partitions, + int messagesPerPartition, + int maxScanRecords, long deadlineNs, Duration pollTimeout ) { - List> messages = new ArrayList<>(); - while (messages.size() < count && System.nanoTime() < deadlineNs) { + PeekCollectionState collection = new PeekCollectionState(partitions, messagesPerPartition); + collectPeekedMessages( + poller, + state -> state.allPartitionQuotasSatisfied() || caughtUpChecker.allPartitionsCaughtUp(), + recordFilter, + collection, + maxScanRecords, + deadlineNs, + pollTimeout + ); + return collection.messages; + } + + private static boolean collectPeekedMessages( + PeekRecordPoller poller, + PeekCollectionCompletionChecker completionChecker, + PeekRecordFilter recordFilter, + PeekCollectionState collection, + int maxScanRecords, + long deadlineNs, + Duration pollTimeout + ) { + while (System.nanoTime() < deadlineNs) { + if (completionChecker.isComplete(collection)) { + return true; + } long remainingNs = deadlineNs - System.nanoTime(); if (remainingNs <= 0) { break; @@ -1125,19 +1243,205 @@ public final class KafkaAgent { : pollTimeout; ConsumerRecords records = poller.poll(timeout); if (records.isEmpty()) { - if (caughtUpChecker.allPartitionsCaughtUp()) { - break; - } continue; } for (ConsumerRecord record : records) { - messages.add(peekedMessageFromRecord(record)); - if (messages.size() >= count) { - break; + if (++collection.scannedRecords > maxScanRecords) { + return false; } + TopicPartition partition = new TopicPartition(record.topic(), record.partition()); + int remaining = collection.remainingByPartition.getOrDefault(partition, 0); + if (remaining <= 0 || !recordFilter.include(record)) { + continue; + } + collection.messages.add(peekedMessageFromRecord(record)); + collection.remainingByPartition.put(partition, remaining - 1); } } - return messages; + return completionChecker.isComplete(collection); + } + + /** + * Starts at the snapshot tail and widens backward when compacted or retained-offset gaps + * leave a partition short of its record quota. + */ + private static PeekCollectionState collectLatestPeekedMessages( + KafkaConsumer consumer, + List partitions, + Map beginningOffsets, + Map initialSeekOffsets, + Map snapshotEndOffsets, + int messagesPerPartition, + int maxScanRecords, + long deadlineNs, + Duration pollTimeout + ) { + Map rangeStartOffsets = new LinkedHashMap<>(initialSeekOffsets); + Map rangeEndOffsets = new LinkedHashMap<>(snapshotEndOffsets); + Map rangeWidths = new LinkedHashMap<>(); + for (TopicPartition partition : partitions) { + long rangeStart = rangeStartOffsets.getOrDefault(partition, 0L); + long rangeEnd = rangeEndOffsets.getOrDefault(partition, rangeStart); + rangeWidths.put(partition, Math.max(1L, rangeEnd - rangeStart)); + } + + PeekCollectionState collection = new PeekCollectionState(partitions, messagesPerPartition); + while (!collection.allPartitionQuotasSatisfied()) { + PeekCollectionCompletionChecker rangeComplete = state -> allPeekPartitionsComplete( + partitions, + state.remainingByPartition, + currentPeekPositions(consumer, partitions), + rangeEndOffsets + ); + if (!collectLatestPeekRange( + timeout -> consumer.poll(timeout), + rangeComplete, + record -> recordIsBeforeEndOffset(record, rangeEndOffsets), + collection, + maxScanRecords, + deadlineNs, + pollTimeout + )) { + collection.incomplete = true; + break; + } + if (collection.allPartitionQuotasSatisfied()) { + break; + } + + boolean expanded = false; + for (TopicPartition partition : partitions) { + if (collection.remainingByPartition.getOrDefault(partition, 0) <= 0) { + continue; + } + long beginningOffset = beginningOffsets.getOrDefault(partition, 0L); + long currentStart = rangeStartOffsets.getOrDefault(partition, beginningOffset); + if (currentStart <= beginningOffset) { + continue; + } + long nextStart = previousLatestPeekStartOffset( + beginningOffset, + currentStart, + rangeWidths.getOrDefault(partition, 1L) + ); + rangeEndOffsets.put(partition, currentStart); + rangeStartOffsets.put(partition, nextStart); + rangeWidths.put(partition, Math.max(1L, currentStart - nextStart)); + consumer.seek(partition, nextStart); + expanded = true; + } + if (!expanded) { + break; + } + } + return collection; + } + + /** + * Scans one backward-expanded range to its end, retaining only each partition's newest + * remaining records. Stopping early would select older records from a widened range. + */ + private static boolean collectLatestPeekRange( + PeekRecordPoller poller, + PeekCollectionCompletionChecker rangeComplete, + PeekRecordFilter recordFilter, + PeekCollectionState collection, + int maxScanRecords, + long deadlineNs, + Duration pollTimeout + ) { + Map>> rangeMessages = new HashMap<>(); + while (System.nanoTime() < deadlineNs) { + if (rangeComplete.isComplete(collection)) { + break; + } + long remainingNs = deadlineNs - System.nanoTime(); + if (remainingNs <= 0) { + break; + } + Duration timeout = pollTimeout.toNanos() > remainingNs + ? Duration.ofNanos(remainingNs) + : pollTimeout; + ConsumerRecords records = poller.poll(timeout); + for (ConsumerRecord record : records) { + if (++collection.scannedRecords > maxScanRecords) { + commitLatestPeekRange(rangeMessages, collection); + return false; + } + TopicPartition partition = new TopicPartition(record.topic(), record.partition()); + int remaining = collection.remainingByPartition.getOrDefault(partition, 0); + if (remaining <= 0 || !recordFilter.include(record)) { + continue; + } + Deque> latestRecords = rangeMessages.computeIfAbsent( + partition, + ignored -> new ArrayDeque<>() + ); + retainLatestPeekRecord(latestRecords, peekedMessageFromRecord(record), remaining); + } + } + boolean complete = rangeComplete.isComplete(collection); + commitLatestPeekRange(rangeMessages, collection); + return complete; + } + + private static void commitLatestPeekRange( + Map>> rangeMessages, + PeekCollectionState collection + ) { + for (Map.Entry>> entry : rangeMessages.entrySet()) { + int retainedCount = entry.getValue().size(); + collection.messages.addAll(entry.getValue()); + collection.remainingByPartition.computeIfPresent( + entry.getKey(), + (ignored, remaining) -> remaining - retainedCount + ); + } + } + + static void retainLatestPeekRecord(Deque records, T record, int maxRecords) { + records.addLast(record); + if (records.size() > maxRecords) { + records.removeFirst(); + } + } + + private static boolean recordIsBeforeEndOffset( + ConsumerRecord record, + Map endOffsets + ) { + Long endOffset = endOffsets.get(new TopicPartition(record.topic(), record.partition())); + return endOffset != null && record.offset() < endOffset; + } + + private static Map currentPeekPositions( + KafkaConsumer consumer, + List partitions + ) { + Map positions = new LinkedHashMap<>(); + for (TopicPartition partition : partitions) { + positions.put(partition, consumer.position(partition)); + } + return positions; + } + + static boolean allPeekPartitionsComplete( + List partitions, + Map remainingByPartition, + Map positions, + Map endOffsets + ) { + for (TopicPartition partition : partitions) { + if (remainingByPartition.getOrDefault(partition, 0) <= 0) { + continue; + } + long endOffset = endOffsets.getOrDefault(partition, 0L); + long position = positions.getOrDefault(partition, 0L); + if (position < endOffset) { + return false; + } + } + return true; } static boolean allPeekPartitionsCaughtUp( @@ -1165,16 +1469,44 @@ public final class KafkaAgent { boolean allPartitionsCaughtUp(); } + @FunctionalInterface + private interface PeekCollectionCompletionChecker { + boolean isComplete(PeekCollectionState collection); + } + + @FunctionalInterface + interface PeekRecordFilter { + boolean include(ConsumerRecord record); + } + + private static final class PeekCollectionState { + private final List> messages = new ArrayList<>(); + private final Map remainingByPartition = new HashMap<>(); + private int scannedRecords; + private boolean incomplete; + + private PeekCollectionState(List partitions, int messagesPerPartition) { + for (TopicPartition partition : partitions) { + remainingByPartition.put(partition, messagesPerPartition); + } + } + + private boolean allPartitionQuotasSatisfied() { + return remainingByPartition.values().stream().allMatch(remaining -> remaining <= 0); + } + } + /** When partition is null, peek across every partition of the topic. */ static List resolvePeekPartitions( KafkaConsumer consumer, String topic, - Integer partition + Integer partition, + Duration timeout ) { if (partition != null) { return resolvePeekPartitions(topic, partition, Collections.emptyList()); } - List infos = consumer.partitionsFor(topic, Duration.ofSeconds(5)); + List infos = consumer.partitionsFor(topic, timeout); if (infos == null || infos.isEmpty()) { return Collections.emptyList(); } @@ -1195,13 +1527,103 @@ public final class KafkaAgent { .collect(Collectors.toList()); } + enum PeekStartPosition { + EARLIEST, + LATEST, + OFFSET, + } + + /** Omitting startPosition preserves the old earliest default (or explicit legacy offset) behavior. */ + static PeekStartPosition peekStartPosition(JsonObject params) { + String value = stringOrNull(params, "startPosition"); + if (value == null) { + return PeekStartPosition.EARLIEST; + } + return switch (value.trim().toLowerCase(Locale.ROOT)) { + case "earliest" -> PeekStartPosition.EARLIEST; + case "latest" -> PeekStartPosition.LATEST; + case "offset" -> PeekStartPosition.OFFSET; + default -> throw new IllegalArgumentException("Unsupported peek startPosition: " + value); + }; + } + + static void validatePeekRequest( + PeekStartPosition startPosition, + boolean explicitStartPosition, + Integer partition, + Long offset + ) { + if (partition != null && partition < 0) { + throw new IllegalArgumentException("partition must be non-negative"); + } + if (!explicitStartPosition) { + // Older clients used offset directly without a startPosition field. + if (offset != null && offset < 0) { + throw new IllegalArgumentException("offset must be non-negative"); + } + return; + } + if (startPosition != PeekStartPosition.OFFSET) { + if (offset != null) { + throw new IllegalArgumentException("offset is only supported when startPosition is offset"); + } + return; + } + if (offset == null) { + throw new IllegalArgumentException("offset is required when startPosition is offset"); + } + if (offset < 0) { + throw new IllegalArgumentException("offset must be non-negative when startPosition is offset"); + } + } + + static Long requestedPeekOffset( + PeekStartPosition startPosition, + Long offset, + boolean legacyOffsetRequest, + long beginningOffset, + long endOffset + ) { + return switch (startPosition) { + case LATEST -> endOffset > beginningOffset ? beginningOffset : null; + case OFFSET -> offset; + case EARLIEST -> legacyOffsetRequest ? offset : beginningOffset; + }; + } + + static int peekScanLimit(int count, int readablePartitionCount) { + int fetchCount = recentPeekFetchCount( + peekMessagesPerPartition(count, readablePartitionCount), readablePartitionCount + ); + if (fetchCount > MAX_PEEK_SCAN_RECORDS) { + throw new IllegalArgumentException( + "Kafka message browse would scan more than " + MAX_PEEK_SCAN_RECORDS + " records" + ); + } + return MAX_PEEK_SCAN_RECORDS; + } + static void sortPeekedMessages(List> messages) { - messages.sort((left, right) -> { + sortPeekedMessages(messages, PeekStartPosition.EARLIEST); + } + + static void sortPeekedMessages(List> messages, PeekStartPosition startPosition) { + if (startPosition == PeekStartPosition.OFFSET) { + messages.sort(Comparator + .comparingLong((Map message) -> + ((Number) message.getOrDefault("offset", 0L)).longValue() + ) + .thenComparingInt(message -> ((Number) message.getOrDefault("partition", 0)).intValue()) + ); + return; + } + + Comparator> comparator = (left, right) -> { long leftTs = ((Number) left.getOrDefault("timestamp", 0L)).longValue(); long rightTs = ((Number) right.getOrDefault("timestamp", 0L)).longValue(); int byTs = Long.compare(leftTs, rightTs); if (byTs != 0) { - return byTs; + return startPosition == PeekStartPosition.LATEST ? -byTs : byTs; } int leftPartition = ((Number) left.getOrDefault("partition", 0)).intValue(); int rightPartition = ((Number) right.getOrDefault("partition", 0)).intValue(); @@ -1212,7 +1634,8 @@ public final class KafkaAgent { long leftOffset = ((Number) left.getOrDefault("offset", 0L)).longValue(); long rightOffset = ((Number) right.getOrDefault("offset", 0L)).longValue(); return Long.compare(leftOffset, rightOffset); - }); + }; + messages.sort(comparator); } private static Map peekedMessageFromRecord(ConsumerRecord record) { @@ -1224,7 +1647,7 @@ public final class KafkaAgent { msg.put("key", record.key()); Map headers = new LinkedHashMap<>(); record.headers().forEach(h -> - headers.put(h.key(), new String(h.value(), StandardCharsets.UTF_8))); + headers.put(h.key(), h.value() == null ? "" : new String(h.value(), StandardCharsets.UTF_8))); msg.put("headers", headers); if (record.value() != null) { msg.put("payloadBase64", Base64.getEncoder().encodeToString(record.value())); @@ -1584,6 +2007,34 @@ public final class KafkaAgent { return requestedOffset; } + /** Splits the requested page quota across partitions. */ + static int peekMessagesPerPartition(int count, int partitionCount) { + int safePartitionCount = Math.max(1, partitionCount); + return (int) (((long) count + safePartitionCount - 1) / safePartitionCount); + } + + static long recentPeekStartOffset(long beginningOffset, long endOffset, int messagesPerPartition) { + return Math.max(beginningOffset, endOffset - messagesPerPartition); + } + + static long previousLatestPeekStartOffset( + long beginningOffset, + long currentStartOffset, + long currentWindowWidth + ) { + long safeWindowWidth = Math.max(1L, currentWindowWidth); + long expandedWindowWidth = safeWindowWidth > Long.MAX_VALUE / 2 + ? Long.MAX_VALUE + : safeWindowWidth * 2; + long distanceToBeginning = currentStartOffset - beginningOffset; + return currentStartOffset - Math.min(distanceToBeginning, expandedWindowWidth); + } + + static int recentPeekFetchCount(int messagesPerPartition, int partitionCount) { + long total = (long) messagesPerPartition * Math.max(1, partitionCount); + return (int) Math.min(Integer.MAX_VALUE, total); + } + private static String stringOrNull(JsonObject object, String key) { JsonElement element = object.get(key); return element == null || element.isJsonNull() ? null : element.getAsString(); 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 c76cfd947f..b34b01bc8d 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 @@ -1,9 +1,11 @@ package com.dbx.agent.kafka; import com.google.gson.JsonObject; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import com.google.gson.JsonParser; @@ -12,8 +14,11 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; +import java.util.ArrayDeque; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Deque; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -302,6 +307,183 @@ class KafkaAgentTest { assertNull(KafkaAgent.normalizePeekOffset(0, 5, 5)); } + @Test + void peekStartPositionDefaultsToEarliestForOlderClients() { + assertEquals(KafkaAgent.PeekStartPosition.EARLIEST, + KafkaAgent.peekStartPosition(new JsonObject())); + } + + @Test + void peekStartPositionRecognizesEveryExplicitMode() { + JsonObject latest = new JsonObject(); + latest.addProperty("startPosition", "latest"); + JsonObject earliest = new JsonObject(); + earliest.addProperty("startPosition", "earliest"); + JsonObject offset = new JsonObject(); + offset.addProperty("startPosition", "offset"); + + assertEquals(KafkaAgent.PeekStartPosition.LATEST, KafkaAgent.peekStartPosition(latest)); + assertEquals(KafkaAgent.PeekStartPosition.EARLIEST, KafkaAgent.peekStartPosition(earliest)); + assertEquals(KafkaAgent.PeekStartPosition.OFFSET, KafkaAgent.peekStartPosition(offset)); + } + + @Test + void peekStartPositionRejectsUnknownValues() { + JsonObject params = new JsonObject(); + params.addProperty("startPosition", "middle"); + + assertThrows(IllegalArgumentException.class, () -> KafkaAgent.peekStartPosition(params)); + } + + @Test + void offsetStartPositionAllowsAllPartitionsButRequiresANonNegativeOffset() { + assertDoesNotThrow(() -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.OFFSET, true, null, 0L)); + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.OFFSET, true, 0, null)); + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.OFFSET, true, -1, 0L)); + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.OFFSET, true, 0, -1L)); + } + + @Test + void nonOffsetStartPositionsRejectAnOffset() { + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.LATEST, true, 0, 7L)); + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.EARLIEST, true, 0, 7L)); + } + + @Test + void latestSkipsEmptyPartitions() { + assertNull(KafkaAgent.requestedPeekOffset( + KafkaAgent.PeekStartPosition.LATEST, null, false, 5L, 5L + )); + } + + @Test + void everyStartPositionRejectsNegativePartitions() { + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.LATEST, true, -1, null)); + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.EARLIEST, true, -1, null)); + } + + @Test + void legacyOffsetWithoutStartPositionKeepsTheExistingReadBehavior() { + KafkaAgent.validatePeekRequest(KafkaAgent.PeekStartPosition.EARLIEST, false, null, 7L); + assertEquals(7L, KafkaAgent.requestedPeekOffset( + KafkaAgent.PeekStartPosition.EARLIEST, 7L, true, 0L, 10L + )); + } + + @Test + void explicitEarliestDoesNotReuseAnOffsetFromAnOlderRequest() { + assertEquals(0L, KafkaAgent.requestedPeekOffset( + KafkaAgent.PeekStartPosition.EARLIEST, 7L, false, 0L, 10L + )); + } + + @Test + void offsetSortUsesPartitionAsADeterministicTieBreaker() { + var messages = new java.util.ArrayList>(); + messages.add(Map.of("partition", 2, "offset", 7L)); + messages.add(Map.of("partition", 1, "offset", 7L)); + messages.add(Map.of("partition", 0, "offset", 8L)); + + KafkaAgent.sortPeekedMessages(messages, KafkaAgent.PeekStartPosition.OFFSET); + + assertEquals(1, messages.get(0).get("partition")); + assertEquals(2, messages.get(1).get("partition")); + assertEquals(8L, messages.get(2).get("offset")); + } + + @Test + void splitsMessageWindowAcrossPartitions() { + assertEquals(4, KafkaAgent.peekMessagesPerPartition(10, 3)); + assertEquals(10, KafkaAgent.peekMessagesPerPartition(10, 1)); + } + + @Test + void startsLatestMessageWindowNearThePartitionEnd() { + assertEquals(90L, KafkaAgent.recentPeekStartOffset(0, 100, 10)); + assertEquals(5L, KafkaAgent.recentPeekStartOffset(5, 8, 10)); + } + + @Test + void boundsThePerPartitionMessageQuotaBeforeTrimmingTheResult() { + assertEquals(12, KafkaAgent.recentPeekFetchCount(4, 3)); + assertEquals(4, KafkaAgent.peekMessagesPerPartition(10, 3)); + } + + @Test + void peekRejectsAWindowThatExceedsTheScanLimit() { + assertThrows(IllegalArgumentException.class, () -> + KafkaAgent.peekScanLimit(100, 1_001)); + } + + @Test + void peekWindowCalculationDoesNotOverflow() { + assertEquals(1, KafkaAgent.peekMessagesPerPartition( + Integer.MAX_VALUE, Integer.MAX_VALUE + )); + } + + @Test + void latestPeekExpandsBackwardAcrossSparseOffsetGaps() { + assertEquals(6L, KafkaAgent.recentPeekStartOffset(0L, 11L, 5)); + assertEquals(0L, KafkaAgent.previousLatestPeekStartOffset(0L, 6L, 5L)); + assertEquals(12L, KafkaAgent.previousLatestPeekStartOffset(0L, 32L, 10L)); + } + + @Test + void latestPeekRetainsTheNewestRecordsFromAnExpandedRange() { + Deque latestOffsets = new ArrayDeque<>(); + for (long offset = 86L; offset <= 95L; offset++) { + KafkaAgent.retainLatestPeekRecord(latestOffsets, offset, 4); + } + + assertEquals(List.of(92L, 93L, 94L, 95L), new ArrayList<>(latestOffsets)); + } + + @Test + void peekCountMustStayWithinTheServiceLimit() { + assertEquals(100, KafkaAgent.validatedPeekCount(100)); + assertThrows(IllegalArgumentException.class, () -> KafkaAgent.validatedPeekCount(0)); + assertThrows(IllegalArgumentException.class, () -> KafkaAgent.validatedPeekCount(101)); + } + + @Test + void peekUsesTheConfiguredConsumerRequestTimeout() { + Properties properties = new Properties(); + properties.put("request.timeout.ms", "1500"); + + assertEquals(1_500, KafkaAgent.peekRequestTimeoutMs(new JsonObject(), properties)); + } + + @Test + void peekRequestTimeoutPrefersTheConnectionOverrideAndRejectsInvalidValues() { + Properties properties = new Properties(); + properties.put("request.timeout.ms", "1500"); + JsonObject connection = new JsonObject(); + connection.addProperty("request_timeout_ms", 2_500); + + assertEquals(2_500, KafkaAgent.peekRequestTimeoutMs(connection, properties)); + + properties.put("request.timeout.ms", "0"); + assertThrows(IllegalArgumentException.class, () -> KafkaAgent.peekRequestTimeoutMs(new JsonObject(), properties)); + } + + @Test + void incompletePeekResultsAreExplicitlyMarked() { + Map partial = KafkaAgent.peekMessagesResult(List.of(), true); + Map complete = KafkaAgent.peekMessagesResult(List.of(), false); + + assertEquals(true, partial.get("incomplete")); + assertEquals(false, complete.get("incomplete")); + } + @Test void resolvePeekPartitionsUsesSinglePartitionWhenSpecified() { var partitions = KafkaAgent.resolvePeekPartitions("events", 2, List.of(0, 1, 2)); @@ -330,6 +512,37 @@ class KafkaAgentTest { assertEquals(20L, messages.get(3).get("timestamp")); } + @Test + void sortPeekedMessagesCanOrderNewestFirst() { + var messages = new java.util.ArrayList>(); + messages.add(Map.of("timestamp", 20L, "partition", 1, "offset", 1L)); + messages.add(Map.of("timestamp", 10L, "partition", 0, "offset", 5L)); + messages.add(Map.of("timestamp", 10L, "partition", 0, "offset", 2L)); + messages.add(Map.of("timestamp", 10L, "partition", 1, "offset", 0L)); + + KafkaAgent.sortPeekedMessages(messages, KafkaAgent.PeekStartPosition.LATEST); + + assertEquals(20L, messages.get(0).get("timestamp")); + assertEquals(0, messages.get(1).get("partition")); + assertEquals(2L, messages.get(1).get("offset")); + assertEquals(5L, messages.get(2).get("offset")); + assertEquals(1, messages.get(3).get("partition")); + } + + @Test + void sortPeekedMessagesOrdersOffsetModeByOffsetAscending() { + var messages = new java.util.ArrayList>(); + messages.add(Map.of("timestamp", 10L, "partition", 0, "offset", 5L)); + messages.add(Map.of("timestamp", 30L, "partition", 0, "offset", 2L)); + messages.add(Map.of("timestamp", 20L, "partition", 0, "offset", 3L)); + + KafkaAgent.sortPeekedMessages(messages, KafkaAgent.PeekStartPosition.OFFSET); + + assertEquals(2L, messages.get(0).get("offset")); + assertEquals(3L, messages.get(1).get("offset")); + assertEquals(5L, messages.get(2).get("offset")); + } + @Test void allPeekPartitionsCaughtUpRequiresEveryPartitionAtEndOffset() { TopicPartition p0 = new TopicPartition("events", 0); @@ -348,6 +561,33 @@ class KafkaAgentTest { )); } + @Test + void peekCompletionStopsAfterEachPartitionSuppliesItsQuota() { + TopicPartition p0 = new TopicPartition("events", 0); + TopicPartition p1 = new TopicPartition("events", 1); + List partitions = List.of(p0, p1); + Map endOffsets = Map.of(p0, 100L, p1, 100L); + + assertTrue(KafkaAgent.allPeekPartitionsComplete( + partitions, + Map.of(p0, 0, p1, 0), + Map.of(p0, 1L, p1, 1L), + endOffsets + )); + assertFalse(KafkaAgent.allPeekPartitionsComplete( + partitions, + Map.of(p0, 0, p1, 1), + Map.of(p0, 1L, p1, 1L), + endOffsets + )); + assertTrue(KafkaAgent.allPeekPartitionsComplete( + partitions, + Map.of(p0, 0, p1, 1), + Map.of(p0, 1L, p1, 100L), + endOffsets + )); + } + @Test void collectPeekedMessagesRetriesAfterEmptyFirstPoll() { TopicPartition tp = new TopicPartition("events", 0); @@ -365,8 +605,11 @@ class KafkaAgentTest { AtomicInteger polls = new AtomicInteger(); List> messages = KafkaAgent.collectPeekedMessages( timeout -> polls.getAndIncrement() == 0 ? ConsumerRecords.empty() : withData, - () -> false, + () -> polls.get() >= 2, + ignored -> true, + List.of(tp), 1, + 1_000, System.nanoTime() + Duration.ofSeconds(5).toNanos(), Duration.ofMillis(1) ); @@ -378,7 +621,8 @@ class KafkaAgentTest { } @Test - void collectPeekedMessagesStopsOnEmptyPollWhenCaughtUp() { + void collectPeekedMessagesDoesNotPollWhenAlreadyCaughtUp() { + TopicPartition tp = new TopicPartition("events", 0); AtomicInteger polls = new AtomicInteger(); List> messages = KafkaAgent.collectPeekedMessages( timeout -> { @@ -386,13 +630,208 @@ class KafkaAgentTest { return ConsumerRecords.empty(); }, () -> true, + record -> true, + List.of(tp), 10, + 1_000, + System.nanoTime() + Duration.ofSeconds(5).toNanos(), + Duration.ofMillis(1) + ); + + assertEquals(0, polls.get()); + assertTrue(messages.isEmpty()); + } + + @Test + void collectPeekedMessagesExcludesRecordsPastTheSnapshotEndOffset() { + TopicPartition tp = new TopicPartition("events", 0); + ConsumerRecord included = new ConsumerRecord<>( + "events", 0, 9L, "before", "before".getBytes(StandardCharsets.UTF_8) + ); + ConsumerRecord excluded = new ConsumerRecord<>( + "events", 0, 10L, "after", "after".getBytes(StandardCharsets.UTF_8) + ); + ConsumerRecords batch = new ConsumerRecords<>(Map.of(tp, List.of(included, excluded))); + AtomicInteger polls = new AtomicInteger(); + AtomicInteger caughtUpChecks = new AtomicInteger(); + + List> messages = KafkaAgent.collectPeekedMessages( + timeout -> { + polls.incrementAndGet(); + return batch; + }, + () -> caughtUpChecks.getAndIncrement() > 0, + record -> record.offset() < 10L, + List.of(tp), + 2, + 1_000, System.nanoTime() + Duration.ofSeconds(5).toNanos(), Duration.ofMillis(1) ); assertEquals(1, polls.get()); - assertTrue(messages.isEmpty()); + assertEquals(1, messages.size()); + assertEquals(9L, messages.get(0).get("offset")); + } + + @Test + void peekCollectsFromEveryPartitionBeforeTrimming() { + TopicPartition p0 = new TopicPartition("events", 0); + TopicPartition p1 = new TopicPartition("events", 1); + ConsumerRecords batch = new ConsumerRecords<>(Map.of( + p0, List.of( + new ConsumerRecord<>("events", 0, 9L, "p0-first", "one".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("events", 0, 10L, "p0-second", "two".getBytes(StandardCharsets.UTF_8)) + ), + p1, List.of( + new ConsumerRecord<>("events", 1, 7L, "p1-first", "three".getBytes(StandardCharsets.UTF_8)) + ) + )); + AtomicInteger polls = new AtomicInteger(); + + List> messages = KafkaAgent.collectPeekedMessages( + timeout -> polls.getAndIncrement() == 0 ? batch : ConsumerRecords.empty(), + () -> polls.get() > 0, + record -> true, + List.of(p0, p1), + 1, + 1_000, + System.nanoTime() + Duration.ofSeconds(5).toNanos(), + Duration.ofMillis(1) + ); + + assertEquals(2, messages.size()); + assertTrue(messages.stream().anyMatch(message -> message.get("partition").equals(0))); + assertTrue(messages.stream().anyMatch(message -> message.get("partition").equals(1))); + } + + @Test + void peekWaitsForEveryPartitionWindowWhenOnePartitionRespondsFirst() { + TopicPartition p0 = new TopicPartition("events", 0); + TopicPartition p1 = new TopicPartition("events", 1); + ConsumerRecords firstPartition = new ConsumerRecords<>(Map.of(p0, List.of( + new ConsumerRecord<>("events", 0, 0L, "p0-first", "one".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("events", 0, 1L, "p0-second", "two".getBytes(StandardCharsets.UTF_8)) + ))); + ConsumerRecords secondPartition = new ConsumerRecords<>(Map.of(p1, List.of( + new ConsumerRecord<>("events", 1, 0L, "p1-first", "three".getBytes(StandardCharsets.UTF_8)) + ))); + AtomicInteger polls = new AtomicInteger(); + + List> messages = KafkaAgent.collectPeekedMessages( + timeout -> polls.getAndIncrement() == 0 ? firstPartition : secondPartition, + () -> polls.get() >= 2, + ignored -> true, + List.of(p0, p1), + 1, + 1_000, + System.nanoTime() + Duration.ofSeconds(5).toNanos(), + Duration.ofMillis(1) + ); + + assertEquals(2, polls.get()); + assertEquals(2, messages.size()); + assertTrue(messages.stream().anyMatch(message -> message.get("partition").equals(0))); + assertTrue(messages.stream().anyMatch(message -> message.get("partition").equals(1))); + } + + @Test + void peekRetainsRecordsReadBeforeTheScanLimit() { + TopicPartition partition = new TopicPartition("events", 0); + ConsumerRecords batch = new ConsumerRecords<>(Map.of(partition, List.of( + new ConsumerRecord<>("events", 0, 9L, "first", "one".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("events", 0, 10L, "second", "two".getBytes(StandardCharsets.UTF_8)) + ))); + + List> messages = KafkaAgent.collectPeekedMessages( + timeout -> batch, + () -> false, + record -> true, + List.of(partition), + 2, + 1, + System.nanoTime() + Duration.ofSeconds(5).toNanos(), + Duration.ofMillis(1) + ); + + assertEquals(1, messages.size()); + assertEquals(9L, messages.get(0).get("offset")); + } + + @Test + void peekCountsSparseOffsetsAsRecordsInsteadOfOffsetWindowWidth() { + TopicPartition partition = new TopicPartition("events", 0); + // A compacted topic can retain these two records while offsets 1..9 are absent. + ConsumerRecords batch = new ConsumerRecords<>(Map.of(partition, List.of( + new ConsumerRecord<>("events", 0, 0L, "first", "one".getBytes(StandardCharsets.UTF_8)), + new ConsumerRecord<>("events", 0, 10L, "second", "two".getBytes(StandardCharsets.UTF_8)) + ))); + AtomicInteger polls = new AtomicInteger(); + + List> messages = KafkaAgent.collectPeekedMessages( + timeout -> polls.getAndIncrement() == 0 ? batch : ConsumerRecords.empty(), + () -> polls.get() > 0, + record -> true, + List.of(partition), + 5, + 1_000, + System.nanoTime() + Duration.ofSeconds(5).toNanos(), + Duration.ofMillis(1) + ); + + assertEquals(2, messages.size()); + assertEquals(0L, messages.get(0).get("offset")); + assertEquals(10L, messages.get(1).get("offset")); + } + + @Test + void peekHandlesKafkaHeadersWithNullValues() { + TopicPartition partition = new TopicPartition("events", 0); + ConsumerRecord record = new ConsumerRecord<>( + "events", 0, 0L, "key", "value".getBytes(StandardCharsets.UTF_8) + ); + record.headers().add("tombstone", null); + ConsumerRecords batch = new ConsumerRecords<>(Map.of(partition, List.of(record))); + AtomicInteger polls = new AtomicInteger(); + + List> messages = KafkaAgent.collectPeekedMessages( + timeout -> { + polls.incrementAndGet(); + return batch; + }, + () -> polls.get() > 0, + ignored -> true, + List.of(partition), + 1, + 1_000, + System.nanoTime() + Duration.ofSeconds(5).toNanos(), + Duration.ofMillis(1) + ); + + assertEquals("", ((Map) messages.get(0).get("headers")).get("tombstone")); + } + + @Test + void incompleteCollectCanBeReturnedWithAnExplicitStatus() { + TopicPartition partition = new TopicPartition("events", 0); + ConsumerRecords batch = new ConsumerRecords<>(Map.of(partition, List.of( + new ConsumerRecord<>("events", 0, 0L, "only", "one".getBytes(StandardCharsets.UTF_8)) + ))); + + List> partial = KafkaAgent.collectPeekedMessages( + timeout -> batch, + () -> false, + record -> true, + List.of(partition), + 5, + 1_000, + System.nanoTime() - 1, + Duration.ofMillis(1) + ); + + assertEquals(0, partial.size()); + Map result = KafkaAgent.peekMessagesResult(partial, true); + assertEquals(true, result.get("incomplete")); } @Test diff --git a/apps/desktop/src/components/mq/MessageBrowser.vue b/apps/desktop/src/components/mq/MessageBrowser.vue new file mode 100644 index 0000000000..15f4d0de21 --- /dev/null +++ b/apps/desktop/src/components/mq/MessageBrowser.vue @@ -0,0 +1,462 @@ + + + + + diff --git a/apps/desktop/src/components/mq/MessageQueryPanel.vue b/apps/desktop/src/components/mq/MessageQueryPanel.vue index 1cb3fedc1e..58a24fce80 100644 --- a/apps/desktop/src/components/mq/MessageQueryPanel.vue +++ b/apps/desktop/src/components/mq/MessageQueryPanel.vue @@ -151,7 +151,8 @@ async function runQuery() { } else { const group = consumerGroup.value.trim() || "__dbx_rocketmq_dlq__"; const peeked = await mqPeekMessages(props.connectionId, topic, group, 64); - queryMessages.value = peeked.map((msg) => rocketMqDisplayFromPeeked(msg, topic.topic)); + const messages = Array.isArray(peeked) ? peeked : peeked.messages; + queryMessages.value = messages.map((msg) => rocketMqDisplayFromPeeked(msg, topic.topic)); } } else if (activeQueryMode.value === "msgId") { const id = queryMsgId.value.trim(); diff --git a/apps/desktop/src/components/mq/MonitoringPanel.vue b/apps/desktop/src/components/mq/MonitoringPanel.vue index 8ee995292c..0ee0d2fffa 100644 --- a/apps/desktop/src/components/mq/MonitoringPanel.vue +++ b/apps/desktop/src/components/mq/MonitoringPanel.vue @@ -8,8 +8,9 @@ import { LineChart } from "echarts/charts"; import { GridComponent, LegendComponent, TooltipComponent } from "echarts/components"; import VChart from "vue-echarts"; import { Activity, AlertTriangle, BarChart3, Boxes, CheckCircle2, Database, Download, Gauge, Hash, HardDrive, Layers3, Loader2, Package, RadioTower, RefreshCw, Send, ShieldCheck, Table2, Upload, Users } from "@lucide/vue"; -import type { MqSystemKind, TopicRef, TopicInfo, TopicStats, BacklogStats, PeekedMessage } from "@/types/mq"; -import { mqGetTopicStats, mqGetBacklog, mqPeekMessages } from "@/lib/backend/api"; +import type { MqSystemKind, TopicRef, TopicInfo, TopicStats, BacklogStats } from "@/types/mq"; +import { mqGetTopicStats, mqGetBacklog } from "@/lib/backend/api"; +import MessageBrowser from "./MessageBrowser.vue"; use([CanvasRenderer, LineChart, GridComponent, LegendComponent, TooltipComponent]); @@ -73,10 +74,6 @@ const error = ref(); const autoRefresh = ref(true); const refreshInterval = ref(5); // seconds const selectedPartitionName = ref(); -const kafkaMessageSql = ref(""); -const kafkaMessageLoading = ref(false); -const kafkaMessageError = ref(); -const kafkaMessages = ref([]); let refreshTimer: number | undefined; const history = ref([]); @@ -221,69 +218,6 @@ function refreshNow() { void loadStats(); } -function defaultKafkaMessageSql(): string { - const topic = props.topic?.shortName; - return topic ? `SELECT * FROM "${topic}" LIMIT 20` : ""; -} - -function parseKafkaMessageSql(sql: string): { topic: string; partition?: number; offset?: number; limit: number } { - const match = sql.trim().match(/^\s*select\s+\*\s+from\s+(?:"([^"]+)"|`([^`]+)`|'([^']+)'|([^\s;]+))(?:\s+partition\s+(\d+))?(?:\s+offset\s+(\d+))?(?:\s+limit\s+(\d+))?\s*;?\s*$/i); - if (!match) { - throw new Error(t("mqMonitoring.sqlSyntaxError")); - } - const topic = match[1] || match[2] || match[3] || match[4] || ""; - const partition = match[5] != null ? Math.max(0, Number(match[5])) : undefined; - const offset = match[6] != null ? Math.max(0, Number(match[6])) : undefined; - const limit = Math.max(1, Math.min(100, Number(match[7] ?? 20))); - return { topic, partition, offset, limit }; -} - -function flatMqMonitorGroupName(): string { - if (props.mqSystemKind === "rocketmq") return "__dbx_rocketmq_monitor__"; - return "__dbx_kafka_monitor__"; -} - -async function runKafkaMessageSql() { - if (!props.tenant || !props.namespace) return; - kafkaMessageLoading.value = true; - kafkaMessageError.value = undefined; - try { - const parsed = parseKafkaMessageSql(kafkaMessageSql.value); - const selected = props.topic && parsed.topic === props.topic.shortName ? props.topic : undefined; - const options: { partition?: number; offset?: number } = {}; - if (parsed.partition != null) options.partition = parsed.partition; - if (parsed.offset != null) options.offset = parsed.offset; - kafkaMessages.value = await mqPeekMessages( - props.connectionId, - { - tenant: props.tenant, - namespace: props.namespace, - topic: parsed.topic, - persistent: selected?.persistent ?? true, - partitioned: selected?.partitioned, - }, - flatMqMonitorGroupName(), - parsed.limit, - options, - ); - } catch (e: unknown) { - kafkaMessageError.value = formatError(e); - } finally { - kafkaMessageLoading.value = false; - } -} - -function kafkaMessagePayload(message: PeekedMessage): string { - return message.payloadText ?? message.payloadBase64; -} - -function formatKafkaMessageTimestamp(value?: string): string { - if (!value) return "-"; - const numeric = Number(value); - if (!Number.isFinite(numeric)) return value; - return new Date(numeric).toLocaleString(); -} - function appendHistoryPoint(statsData: TopicStats, backlogData?: BacklogStats) { const point: MetricPoint = { time: new Date().toLocaleTimeString(), @@ -452,9 +386,6 @@ watch( () => { history.value = []; selectedPartitionName.value = undefined; - kafkaMessageSql.value = defaultKafkaMessageSql(); - kafkaMessageError.value = undefined; - kafkaMessages.value = []; void loadStats(); startAutoRefresh(); }, @@ -515,16 +446,16 @@ onUnmounted(() => { +
+ + {{ error }} +
+
{{ t("mqMonitoring.selectTopicFirst") }}
-
- - {{ error }} -
-
{{ t("mqMonitoring.loadingStats") }} @@ -698,36 +629,7 @@ onUnmounted(() => {
-
-

{{ t("mqMonitoring.kafkaMessageQuery") }}

- -
-