fix(kafka): improve message browsing and topic navigation

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

@@ -1140,8 +1140,12 @@ public final class KafkaAgent {
                 return peekMessagesResult(Collections.emptyList(), false);
             }
 
-            int messagesPerPartition = peekMessagesPerPartition(count, readablePartitions.size());
-            int scanLimit = peekScanLimit(count, readablePartitions.size());
+            int messagesPerPartition = startPosition == PeekStartPosition.LATEST
+                ? latestPeekMessagesPerPartition(count, readablePartitions.size())
+                : peekMessagesPerPartition(count, readablePartitions.size());
+            int scanLimit = peekScanLimit(count, readablePartitions.size(), startPosition);
+            boolean latestBudgetLimited = startPosition == PeekStartPosition.LATEST
+                && latestPeekBudgetLimited(count, readablePartitions.size());
             Map<TopicPartition, Long> snapshotEndOffsets = new LinkedHashMap<>();
             if (startPosition == PeekStartPosition.LATEST) {
                 for (TopicPartition tp : readablePartitions) {
@@ -1176,6 +1180,7 @@ public final class KafkaAgent {
                     deadlineNs,
                     pollTimeout
                 );
+                collection.incomplete = collection.incomplete || latestBudgetLimited;
             } else {
                 PeekCollectionCompletionChecker snapshotComplete = state -> allPeekPartitionsComplete(
                     readablePartitions,
@@ -1194,11 +1199,9 @@ public final class KafkaAgent {
                     pollTimeout
                 );
             }
-            List<Map<String, Object>> messages = collection.messages;
-            sortPeekedMessages(messages, startPosition);
-            if (messages.size() > count) {
-                messages = new ArrayList<>(messages.subList(0, count));
-            }
+            List<Map<String, Object>> messages = sortAndLimitPeekedMessages(
+                collection.messages, count, startPosition
+            );
             return peekMessagesResult(messages, collection.incomplete);
         }
     }
@@ -1311,9 +1314,10 @@ public final class KafkaAgent {
                 continue;
             }
             for (ConsumerRecord<String, byte[]> record : records) {
-                if (++collection.scannedRecords > maxScanRecords) {
+                if (collection.scannedRecords >= maxScanRecords) {
                     return false;
                 }
+                collection.scannedRecords++;
                 TopicPartition partition = new TopicPartition(record.topic(), record.partition());
                 int remaining = collection.remainingByPartition.getOrDefault(partition, 0);
                 if (remaining <= 0 || !recordFilter.include(record)) {
@@ -1429,10 +1433,11 @@ public final class KafkaAgent {
                 : pollTimeout;
             ConsumerRecords<String, byte[]> records = poller.poll(timeout);
             for (ConsumerRecord<String, byte[]> record : records) {
-                if (++collection.scannedRecords > maxScanRecords) {
+                if (collection.scannedRecords >= maxScanRecords) {
                     commitLatestPeekRange(rangeMessages, collection);
                     return false;
                 }
+                collection.scannedRecords++;
                 TopicPartition partition = new TopicPartition(record.topic(), record.partition());
                 int remaining = collection.remainingByPartition.getOrDefault(partition, 0);
                 if (remaining <= 0 || !recordFilter.include(record)) {
@@ -1568,19 +1573,25 @@ public final class KafkaAgent {
         Integer partition,
         Duration timeout
     ) {
-        if (partition != null) {
-            return resolvePeekPartitions(topic, partition, Collections.emptyList());
-        }
         List<PartitionInfo> infos = consumer.partitionsFor(topic, timeout);
         if (infos == null || infos.isEmpty()) {
             return Collections.emptyList();
         }
         List<Integer> available = infos.stream().map(PartitionInfo::partition).collect(Collectors.toList());
-        return resolvePeekPartitions(topic, null, available);
+        return resolvePeekPartitions(topic, partition, available);
     }
 
     static List<TopicPartition> resolvePeekPartitions(String topic, Integer partition, List<Integer> availablePartitions) {
         if (partition != null) {
+            if (availablePartitions == null || !availablePartitions.contains(partition)) {
+                String available = availablePartitions == null || availablePartitions.isEmpty()
+                    ? "none"
+                    : availablePartitions.stream().sorted().map(String::valueOf).collect(Collectors.joining(", "));
+                throw new IllegalArgumentException(
+                    "Kafka partition " + partition + " does not exist for topic '" + topic
+                        + "'. Available partitions: " + available
+                );
+            }
             return Collections.singletonList(new TopicPartition(topic, partition));
         }
         if (availablePartitions == null || availablePartitions.isEmpty()) {
@@ -1657,17 +1668,52 @@ public final class KafkaAgent {
     }
 
     static int peekScanLimit(int count, int readablePartitionCount) {
-        int fetchCount = recentPeekFetchCount(
-            peekMessagesPerPartition(count, readablePartitionCount), readablePartitionCount
-        );
-        if (fetchCount > MAX_PEEK_SCAN_RECORDS) {
+        return peekScanLimit(count, readablePartitionCount, PeekStartPosition.EARLIEST);
+    }
+
+    static int peekScanLimit(int count, int readablePartitionCount, PeekStartPosition startPosition) {
+        if (startPosition == PeekStartPosition.LATEST
+            && readablePartitionCount > MAX_PEEK_SCAN_RECORDS) {
             throw new IllegalArgumentException(
-                "Kafka message browse would scan more than " + MAX_PEEK_SCAN_RECORDS + " records"
+                "Kafka topic has more than " + MAX_PEEK_SCAN_RECORDS
+                    + " readable partitions; select a partition to browse latest messages"
             );
         }
         return MAX_PEEK_SCAN_RECORDS;
     }
 
+    /**
+     * Latest is a topic-level query. Below the scan budget, every partition contributes the
+     * requested count so the global merge is exact. Above it, the fixed budget is shared across
+     * partitions and the response is marked incomplete.
+     */
+    static int latestPeekMessagesPerPartition(int count, int readablePartitionCount) {
+        int safePartitionCount = Math.max(1, readablePartitionCount);
+        if (safePartitionCount > MAX_PEEK_SCAN_RECORDS) {
+            throw new IllegalArgumentException(
+                "Kafka topic has more than " + MAX_PEEK_SCAN_RECORDS
+                    + " readable partitions; select a partition to browse latest messages"
+            );
+        }
+        return Math.min(count, MAX_PEEK_SCAN_RECORDS / safePartitionCount);
+    }
+
+    static boolean latestPeekBudgetLimited(int count, int readablePartitionCount) {
+        return latestPeekMessagesPerPartition(count, readablePartitionCount) < count;
+    }
+
+    static List<Map<String, Object>> sortAndLimitPeekedMessages(
+        List<Map<String, Object>> messages,
+        int count,
+        PeekStartPosition startPosition
+    ) {
+        sortPeekedMessages(messages, startPosition);
+        if (messages.size() <= count) {
+            return messages;
+        }
+        return new ArrayList<>(messages.subList(0, count));
+    }
+
     static void sortPeekedMessages(List<Map<String, Object>> messages) {
         sortPeekedMessages(messages, PeekStartPosition.EARLIEST);
     }
@@ -1698,7 +1744,9 @@ 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);
+            return startPosition == PeekStartPosition.LATEST
+                ? Long.compare(rightOffset, leftOffset)
+                : Long.compare(leftOffset, rightOffset);
         };
         messages.sort(comparator);
     }
@@ -2142,11 +2190,6 @@ public final class KafkaAgent {
         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();

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

@@ -568,15 +568,23 @@ class KafkaAgentTest {
     }
 
     @Test
-    void boundsThePerPartitionMessageQuotaBeforeTrimmingTheResult() {
-        assertEquals(12, KafkaAgent.recentPeekFetchCount(4, 3));
+    void latestPeekReportsWhetherTheGlobalBudgetCanCoverEveryPartitionQuota() {
+        assertFalse(KafkaAgent.latestPeekBudgetLimited(20, 50));
+        assertTrue(KafkaAgent.latestPeekBudgetLimited(20, 51));
+        assertTrue(KafkaAgent.latestPeekBudgetLimited(100, 11));
         assertEquals(4, KafkaAgent.peekMessagesPerPartition(10, 3));
     }
 
     @Test
-    void peekRejectsAWindowThatExceedsTheScanLimit() {
-        assertThrows(IllegalArgumentException.class, () ->
-            KafkaAgent.peekScanLimit(100, 1_001));
+    void latestPeekSharesTheGlobalScanBudgetAcrossManyPartitions() {
+        assertEquals(20, KafkaAgent.latestPeekMessagesPerPartition(20, 3));
+        assertEquals(19, KafkaAgent.latestPeekMessagesPerPartition(20, 51));
+        assertEquals(16, KafkaAgent.latestPeekMessagesPerPartition(20, 60));
+        assertEquals(10, KafkaAgent.latestPeekMessagesPerPartition(20, 100));
+        assertEquals(90, KafkaAgent.latestPeekMessagesPerPartition(100, 11));
+        assertEquals(1_000, KafkaAgent.peekScanLimit(
+            100, 11, KafkaAgent.PeekStartPosition.LATEST
+        ));
     }
 
     @Test
@@ -648,6 +656,25 @@ class KafkaAgentTest {
         assertEquals("events", partitions.get(0).topic());
     }
 
+    @Test
+    void resolvePeekPartitionsRejectsMissingPartitionBeforeOffsetLookup() {
+        IllegalArgumentException error = assertThrows(IllegalArgumentException.class, () ->
+            KafkaAgent.resolvePeekPartitions("events", 5, List.of(0, 1, 2))
+        );
+
+        assertEquals(
+            "Kafka partition 5 does not exist for topic 'events'. Available partitions: 0, 1, 2",
+            error.getMessage()
+        );
+    }
+
+    @Test
+    void resolvePeekPartitionsKeepsRequestedPartitionAfterMetadataLookup() {
+        var partitions = KafkaAgent.resolvePeekPartitions("events", 1, List.of(0, 1, 2));
+
+        assertEquals(List.of(1), partitions.stream().map(TopicPartition::partition).toList());
+    }
+
     @Test
     void resolvePeekPartitionsUsesAllPartitionsWhenUnspecified() {
         var partitions = KafkaAgent.resolvePeekPartitions("events", null, List.of(2, 0, 1));
@@ -680,11 +707,47 @@ class KafkaAgentTest {
 
         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(5L, messages.get(1).get("offset"));
+        assertEquals(2L, messages.get(2).get("offset"));
         assertEquals(1, messages.get(3).get("partition"));
     }
 
+    @Test
+    void latestPeekKeepsTheFullQuotaWhenItFitsTheGlobalScanBudget() {
+        assertEquals(1_000, KafkaAgent.peekScanLimit(
+            100, 10, KafkaAgent.PeekStartPosition.LATEST
+        ));
+        assertEquals(100, KafkaAgent.latestPeekMessagesPerPartition(100, 10));
+        assertEquals(20, KafkaAgent.latestPeekMessagesPerPartition(20, 50));
+    }
+
+    @Test
+    void latestPeekRejectsTopicsWhosePartitionCountAloneExceedsTheScanBudget() {
+        IllegalArgumentException error = assertThrows(IllegalArgumentException.class, () ->
+            KafkaAgent.latestPeekMessagesPerPartition(20, 1_001)
+        );
+
+        assertTrue(error.getMessage().contains("select a partition"));
+    }
+
+    @Test
+    void latestPeekGloballyMergesInterleavedPartitionTimelinesBeforeLimiting() {
+        var messages = new java.util.ArrayList<Map<String, Object>>();
+        messages.add(Map.of("timestamp", 101L, "partition", 0, "offset", 8L));
+        messages.add(Map.of("timestamp", 105L, "partition", 1, "offset", 3L));
+        messages.add(Map.of("timestamp", 103L, "partition", 2, "offset", 9L));
+        messages.add(Map.of("timestamp", 104L, "partition", 0, "offset", 9L));
+        messages.add(Map.of("timestamp", 102L, "partition", 1, "offset", 2L));
+
+        List<Map<String, Object>> latest = KafkaAgent.sortAndLimitPeekedMessages(
+            messages, 3, KafkaAgent.PeekStartPosition.LATEST
+        );
+
+        assertEquals(List.of(105L, 104L, 103L), latest.stream()
+            .map(message -> ((Number) message.get("timestamp")).longValue())
+            .toList());
+    }
+
     @Test
     void sortPeekedMessagesOrdersOffsetModeByOffsetAscending() {
         var messages = new java.util.ArrayList<Map<String, Object>>();

apps/desktop/src/components/mq/KafkaMessagesPanel.vue

@@ -0,0 +1,220 @@
+<script setup lang="ts">
+import { computed, ref, watch } from "vue";
+import { useI18n } from "vue-i18n";
+import type { TopicInfo, TopicRef, TopicStats } from "@/types/mq";
+import { mqGetTopicStats } from "@/lib/backend/api";
+import { formatError } from "@/lib/backend/errorUtils";
+import { extractKafkaPartitionRows } from "@/lib/mq/kafkaTopicStats";
+import RocketMqTopicSelect from "./shared/RocketMqTopicSelect.vue";
+import MessageBrowser from "./MessageBrowser.vue";
+import SendMessagePanel from "./SendMessagePanel.vue";
+
+interface Props {
+  connectionId: string;
+  tenant?: string;
+  namespace?: string;
+  topic?: TopicInfo;
+  readOnly?: boolean;
+  canSendMessage?: boolean;
+}
+
+const props = defineProps<Props>();
+const emit = defineEmits<{
+  topicSelected: [topic: TopicInfo | undefined];
+}>();
+const { t } = useI18n();
+
+const topicSelectRef = ref<InstanceType<typeof RocketMqTopicSelect>>();
+const topics = ref<TopicInfo[]>([]);
+const topicName = ref(props.topic?.shortName ?? "");
+const stats = ref<TopicStats>();
+const statsLoading = ref(false);
+const statsError = ref<string>();
+let statsRequestVersion = 0;
+
+const selectedTopic = computed(() => topics.value.find((item) => item.shortName === topicName.value));
+const selectedTopicRef = computed<TopicRef | null>(() => {
+  const name = topicName.value.trim();
+  if (!name || !props.tenant || !props.namespace) return null;
+  return {
+    tenant: props.tenant,
+    namespace: props.namespace,
+    topic: name,
+    persistent: selectedTopic.value?.persistent ?? true,
+    partitioned: selectedTopic.value?.partitioned,
+  };
+});
+const partitionRows = computed(() => extractKafkaPartitionRows(stats.value?.raw));
+
+function handleTopicsLoaded(loaded: TopicInfo[]) {
+  topics.value = loaded;
+  if (topicName.value && loaded.some((item) => item.shortName === topicName.value)) {
+    return;
+  }
+  if (props.topic && loaded.some((item) => item.shortName === props.topic?.shortName)) {
+    topicName.value = props.topic.shortName;
+  } else if (loaded.length === 1) {
+    topicName.value = loaded[0].shortName;
+  } else {
+    topicName.value = "";
+  }
+}
+
+async function loadStats() {
+  const topic = selectedTopicRef.value;
+  const requestVersion = ++statsRequestVersion;
+  stats.value = undefined;
+  statsError.value = undefined;
+  if (!topic) return;
+  statsLoading.value = true;
+  try {
+    const result = await mqGetTopicStats(props.connectionId, topic);
+    if (requestVersion === statsRequestVersion) stats.value = result;
+  } catch (cause: unknown) {
+    if (requestVersion === statsRequestVersion) statsError.value = formatError(cause);
+  } finally {
+    if (requestVersion === statsRequestVersion) statsLoading.value = false;
+  }
+}
+
+watch([() => props.connectionId, () => props.tenant, () => props.namespace, topicName], () => void loadStats(), { immediate: true });
+watch([() => props.connectionId, () => props.tenant, () => props.namespace], () => {
+  topics.value = [];
+  topicName.value = props.topic?.shortName ?? "";
+});
+watch(
+  () => props.topic?.shortName,
+  (name) => {
+    if (name) topicName.value = name;
+  },
+);
+watch(selectedTopic, (topic) => emit("topicSelected", topic));
+</script>
+
+<template>
+  <div class="kafka-messages-panel">
+    <div class="panel-toolbar">
+      <h3>{{ t("mqMessages.queryTitle") }}</h3>
+      <button type="button" class="btn-secondary" :disabled="!tenant || !namespace" @click="topicSelectRef?.loadTopics()">
+        {{ t("mqMessages.refreshTopicList") }}
+      </button>
+    </div>
+
+    <div v-if="!tenant || !namespace" class="panel-placeholder">{{ t("mqMessages.selectNamespaceOrTopicFirst") }}</div>
+    <div v-else class="kafka-messages-content">
+      <section class="kafka-topic-section">
+        <label>
+          <span>{{ t("mqMessages.queryTopicLabel") }}</span>
+          <RocketMqTopicSelect ref="topicSelectRef" v-model="topicName" :connection-id="connectionId" :tenant="tenant" :namespace="namespace" :show-type-filter="false" show-hint @loaded="handleTopicsLoaded" />
+        </label>
+      </section>
+
+      <section v-if="selectedTopicRef" class="partition-overview" data-testid="kafka-partition-overview">
+        <div class="section-heading">
+          <h4>{{ t("mqMonitoring.kafkaPartitionDetails") }}</h4>
+          <button type="button" class="btn-sm" :disabled="statsLoading" @click="loadStats">{{ t("common.refresh") }}</button>
+        </div>
+        <div v-if="statsError" class="panel-error">{{ statsError }}</div>
+        <div v-else-if="statsLoading" class="section-empty">{{ t("common.loading") }}</div>
+        <div v-else-if="!partitionRows.length" class="section-empty">{{ t("mqMonitoring.noKafkaPartitionMetrics") }}</div>
+        <div v-else class="partition-table-wrap">
+          <table class="partition-table">
+            <thead>
+              <tr>
+                <th>{{ t("mqMonitoring.tablePartition") }}</th>
+                <th>{{ t("mqMonitoring.tableBeginOffset") }}</th>
+                <th>{{ t("mqMonitoring.tableLogEndOffset") }}</th>
+                <th>{{ t("mqMonitoring.tableMessageCount") }}</th>
+              </tr>
+            </thead>
+            <tbody>
+              <tr v-for="row in partitionRows" :key="row.partition">
+                <td>{{ row.partition }}</td>
+                <td>{{ row.beginOffset.toLocaleString() }}</td>
+                <td>{{ row.endOffset.toLocaleString() }}</td>
+                <td>{{ row.messageCount.toLocaleString() }}</td>
+              </tr>
+            </tbody>
+          </table>
+        </div>
+      </section>
+
+      <MessageBrowser :connection-id="connectionId" :topic="selectedTopicRef" mq-system-kind="kafka" />
+      <SendMessagePanel v-if="canSendMessage && selectedTopic" :connection-id="connectionId" :tenant="tenant" :namespace="namespace" :topic="selectedTopic" :read-only="readOnly" mq-system-kind="kafka" is-flat-mq-cluster :supports-peek-messages="false" fixed-topic embedded />
+    </div>
+  </div>
+</template>
+
+<style scoped>
+@import "./shared/mqPanel.css";
+
+.kafka-messages-panel,
+.kafka-messages-content {
+  display: flex;
+  flex-direction: column;
+  gap: 14px;
+}
+
+.kafka-messages-panel {
+  padding: 16px;
+}
+
+.kafka-topic-section,
+.partition-overview {
+  padding: 14px;
+  border: 1px solid var(--color-border);
+  border-radius: var(--dbx-radius-fixed-6);
+  background: var(--color-background-secondary);
+}
+
+.kafka-topic-section label {
+  display: flex;
+  flex-direction: column;
+  gap: 6px;
+  color: var(--color-text-secondary);
+  font-size: 12px;
+  font-weight: 500;
+}
+
+.section-heading {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 12px;
+  margin-bottom: 10px;
+}
+
+.section-heading h4 {
+  margin: 0;
+  font-size: 14px;
+}
+
+.partition-table-wrap {
+  overflow: auto;
+}
+
+.partition-table {
+  width: 100%;
+  border-collapse: collapse;
+  font-size: 12px;
+}
+
+.partition-table th,
+.partition-table td {
+  padding: 8px 10px;
+  border-bottom: 1px solid var(--color-border);
+  text-align: left;
+}
+
+.partition-table th {
+  color: var(--color-text-secondary);
+  font-weight: 600;
+}
+
+.section-empty {
+  padding: 14px;
+  color: var(--color-text-tertiary);
+  text-align: center;
+  font-size: 13px;
+}
+</style>

apps/desktop/src/components/mq/MessageBrowser.vue

@@ -144,10 +144,19 @@ function formatMessageTimestamp(value?: string): string {
   return new Date(numeric).toLocaleString();
 }
 
-watch([() => props.connectionId, () => props.mqSystemKind, () => JSON.stringify(props.topic ?? null)], () => {
+watch([() => props.connectionId, () => props.mqSystemKind], () => {
   invalidateMessageRequest();
 });
 
+watch(
+  () => JSON.stringify(props.topic ?? null),
+  () => {
+    partition.value = "";
+    offset.value = "";
+    invalidateMessageRequest();
+  },
+);
+
 watch(kafkaStartPosition, () => {
   // Keep offset values for switching back, but never retain results from another start mode.
   invalidateMessageRequest();
@@ -184,7 +193,7 @@ watch(kafkaStartPosition, () => {
         <span>{{ t("mqMessages.count") }}</span>
         <input v-model.number="count" data-testid="peek-count" type="number" min="1" max="100" :disabled="loading" />
       </label>
-      <label v-if="isKafka">
+      <label v-if="isKafka && !isMonitoring">
         <span>{{ t("mqMessages.startPosition") }}</span>
         <Select v-model="kafkaStartPosition" :disabled="loading">
           <SelectTrigger data-testid="kafka-peek-start-position" class="message-browser-start-position">
@@ -201,7 +210,7 @@ watch(kafkaStartPosition, () => {
         <span>{{ t("mqMessages.partition") }}</span>
         <input v-model="partition" data-testid="kafka-peek-partition" type="number" min="0" :placeholder="t('mqMessages.partitionPlaceholderAll')" :disabled="loading" />
       </label>
-      <label v-if="isKafkaOffsetMode">
+      <label v-if="isKafkaOffsetMode && !isMonitoring">
         <span>{{ t("mqMessages.offset") }}</span>
         <input v-model="offset" data-testid="kafka-peek-offset" type="number" min="0" :placeholder="t('mqMessages.offsetPlaceholderRequired')" :disabled="loading" />
       </label>

apps/desktop/src/components/mq/MonitoringPanel.vue

@@ -10,6 +10,7 @@ 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 } from "@/types/mq";
 import { mqGetTopicStats, mqGetBacklog } from "@/lib/backend/api";
+import { extractKafkaPartitionRows, isKafkaStatsPayload, type KafkaPartitionStatsRow } from "@/lib/mq/kafkaTopicStats";
 import MessageBrowser from "./MessageBrowser.vue";
 
 use([CanvasRenderer, LineChart, GridComponent, LegendComponent, TooltipComponent]);
@@ -54,17 +55,10 @@ interface RocketMqPartitionStatsRow {
   messageCount: number;
 }
 
-interface KafkaPartitionStatsRow {
-  partition: number;
-  beginOffset: number;
-  endOffset: number;
-  messageCount: number;
-  leader: number;
-  replicas: number[];
-  isr: number[];
-}
-
 const props = defineProps<Props>();
+const emit = defineEmits<{
+  "navigate-tab": [payload: { tab: "messages"; topic?: TopicInfo }];
+}>();
 const { t } = useI18n();
 
 const stats = ref<TopicStats>();
@@ -281,26 +275,6 @@ function extractRocketMqPartitionRows(raw: unknown): RocketMqPartitionStatsRow[]
     .sort((a, b) => a.partition - b.partition || a.brokerName.localeCompare(b.brokerName));
 }
 
-function extractKafkaPartitionRows(raw: unknown): KafkaPartitionStatsRow[] {
-  return arrayObjects(objectRecord(raw).partitionStats)
-    .map((body) => ({
-      partition: numberField(body.partition) ?? 0,
-      beginOffset: numberField(body.beginOffset) ?? 0,
-      endOffset: numberField(body.endOffset) ?? 0,
-      messageCount: numberField(body.messageCount) ?? 0,
-      leader: numberField(body.leader) ?? -1,
-      replicas: numberArrayField(body.replicas),
-      isr: numberArrayField(body.isr),
-    }))
-    .sort((a, b) => a.partition - b.partition);
-}
-
-function isKafkaStatsPayload(raw: unknown): boolean {
-  if (props.mqSystemKind === "rocketmq") return false;
-  const root = objectRecord(raw);
-  return Array.isArray(root.partitionStats) || (numberField(root.partitions) !== undefined && numberField(root.replicationFactor) !== undefined && numberField(root.totalMessages) !== undefined);
-}
-
 function isKafkaPartitionHealthy(row: KafkaPartitionStatsRow): boolean {
   return row.leader >= 0 && (row.replicas.length === 0 || row.isr.length >= row.replicas.length);
 }
@@ -336,10 +310,6 @@ function numberField(value: unknown): number | undefined {
   return typeof value === "number" && Number.isFinite(value) ? value : undefined;
 }
 
-function numberArrayField(value: unknown): number[] {
-  return Array.isArray(value) ? value.map(numberField).filter((item): item is number => item !== undefined) : [];
-}
-
 function stringField(value: unknown): string {
   return typeof value === "string" ? value : "";
 }
@@ -629,6 +599,9 @@ onUnmounted(() => {
       </div>
 
       <div class="stats-section">
+        <div class="monitoring-message-heading">
+          <button type="button" class="btn-sm" @click="emit('navigate-tab', { tab: 'messages', topic })">{{ t("mqMessages.queryTitle") }}</button>
+        </div>
         <MessageBrowser :connection-id="connectionId" :topic="getTopicRef()" mq-system-kind="kafka" appearance="monitoring" />
       </div>
     </div>
@@ -1494,6 +1467,14 @@ onUnmounted(() => {
   color: var(--monitor-muted);
 }
 
+.monitoring-message-heading {
+  display: flex;
+  align-items: center;
+  justify-content: flex-end;
+  gap: 12px;
+  margin-bottom: 12px;
+}
+
 @keyframes monitor-spin {
   to {
     transform: rotate(360deg);

apps/desktop/src/components/mq/MqAdminConsole.vue

@@ -35,6 +35,7 @@ import MessageTracePanel from "./MessageTracePanel.vue";
 import RocketMqMessagesPanel from "./RocketMqMessagesPanel.vue";
 import SendMessagePanel from "./SendMessagePanel.vue";
 import MessageQueryPanel from "./MessageQueryPanel.vue";
+import KafkaMessagesPanel from "./KafkaMessagesPanel.vue";
 import BrokerPanel from "./BrokerPanel.vue";
 import RabbitMqClientsPanel from "./rabbitmq/RabbitMqClientsPanel.vue";
 import RabbitMqPermissionsPanel from "./rabbitmq/RabbitMqPermissionsPanel.vue";
@@ -103,6 +104,8 @@ const mqSystemKind = computed<MqSystemKind | undefined>(() => clusterInfo.value?
 const isFlatMqCluster = computed(() => isFlatMqSystemKind(mqSystemKind.value));
 const isRabbitMqCluster = computed(() => mqSystemKind.value === "rabbitmq");
 const isRocketMqCluster = computed(() => mqSystemKind.value === "rocketmq");
+const isKafkaCluster = computed(() => mqSystemKind.value === "kafka");
+const canBrowseKafkaMessages = computed(() => isKafkaCluster.value && canPeekMessages.value);
 const rocketmqClusterLabel = computed(() => {
   if (!isRocketMqCluster.value) return undefined;
   const fromOptions = clusterOptions.value[0];
@@ -473,7 +476,7 @@ onMounted(async () => {
         @subscription-selected="handleSubscriptionSelected"
       />
       <RabbitMqMonitoringPanel v-else-if="activeTab === 'monitoring' && isRabbitMqCluster && canClusterMonitor" :connection-id="connectionId" />
-      <MonitoringPanel v-else-if="activeTab === 'monitoring'" :connection-id="connectionId" :topic="selectedTopic" :tenant="effectiveTenant" :namespace="effectiveNamespace" :mq-system-kind="mqSystemKind" />
+      <MonitoringPanel v-else-if="activeTab === 'monitoring'" :connection-id="connectionId" :topic="selectedTopic" :tenant="effectiveTenant" :namespace="effectiveNamespace" :mq-system-kind="mqSystemKind" @navigate-tab="handleNavigateTab" />
       <RabbitMqClientsPanel v-else-if="activeTab === 'clients' && isRabbitMqCluster && canManageClientConnections" :connection-id="connectionId" :namespace="effectiveNamespace" :read-only="readOnly" />
       <ProducerConsumerPanel
         v-else-if="activeTab === 'clients'"
@@ -510,6 +513,16 @@ onMounted(async () => {
         :prefer-dlq-topic="preferDlqTopic"
       />
       <MessageTracePanel v-else-if="activeTab === 'trace' && isRocketMqCluster && canMessageTrace" :connection-id="connectionId" :tenant="effectiveTenant" :namespace="effectiveNamespace" :topic="selectedTopic" :read-only="readOnly" :mq-system-kind="mqSystemKind" />
+      <KafkaMessagesPanel
+        v-else-if="activeTab === 'messages' && canBrowseKafkaMessages"
+        :connection-id="connectionId"
+        :tenant="effectiveTenant"
+        :namespace="effectiveNamespace"
+        :topic="selectedTopic"
+        :read-only="readOnly"
+        :can-send-message="canSendMessage"
+        @topic-selected="handleProducerTopicSelected"
+      />
       <MessageQueryPanel v-else-if="activeTab === 'messages' && canMessageQuery && !isRocketMqCluster" :connection-id="connectionId" :tenant="effectiveTenant" :namespace="effectiveNamespace" :topic="selectedTopic" :read-only="readOnly" :mq-system-kind="mqSystemKind" />
       <SendMessagePanel
         v-else-if="activeTab === 'messages' && canSendMessage && !isRocketMqCluster && !canMessageQuery"

apps/desktop/src/components/mq/ProducerConsumerPanel.vue

@@ -5,6 +5,7 @@ import { computed, ref, watch } from "vue";
 import { useI18n } from "vue-i18n";
 import type { ConsumerInfo, ProducerInfo, SubscriptionInfo, TopicInfo, TopicRef, TopicStats, MqSystemKind } from "@/types/mq";
 import { mqGetTopicStats, mqListConsumers, mqListProducers, mqListSubscriptions, mqUnloadTopic } from "@/lib/backend/api";
+import { extractKafkaPartitionRows } from "@/lib/mq/kafkaTopicStats";
 import { useMqMutationGuard } from "@/composables/useMqMutationGuard";
 import RocketMqTopicSelect from "./shared/RocketMqTopicSelect.vue";
 
@@ -41,16 +42,6 @@ interface PartitionClientRow {
   subscriptions: SubscriptionInfo[];
 }
 
-interface KafkaPartitionRow {
-  partition: number;
-  beginOffset: number;
-  endOffset: number;
-  messageCount: number;
-  leader: number;
-  replicas: number[];
-  isr: number[];
-}
-
 const producers = ref<ProducerInfo[]>([]);
 const subscriptions = ref<SubscriptionInfo[]>([]);
 const stats = ref<TopicStats>();
@@ -437,20 +428,6 @@ function extractPartitionClientRows(raw: unknown): PartitionClientRow[] {
   });
 }
 
-function extractKafkaPartitionRows(raw: unknown): KafkaPartitionRow[] {
-  return arrayObjects(objectRecord(raw).partitionStats)
-    .map((body) => ({
-      partition: numberField(body.partition) ?? 0,
-      beginOffset: numberField(body.beginOffset) ?? 0,
-      endOffset: numberField(body.endOffset) ?? 0,
-      messageCount: numberField(body.messageCount) ?? 0,
-      leader: numberField(body.leader) ?? -1,
-      replicas: numberArrayField(body.replicas),
-      isr: numberArrayField(body.isr),
-    }))
-    .sort((a, b) => a.partition - b.partition);
-}
-
 function mergeSubscriptionOptions(base: SubscriptionInfo[], partitions: PartitionClientRow[]): SubscriptionInfo[] {
   const byName = new Map<string, SubscriptionInfo>();
   for (const sub of base) {
@@ -531,10 +508,6 @@ function numberField(value: unknown): number | undefined {
   return undefined;
 }
 
-function numberArrayField(value: unknown): number[] {
-  return Array.isArray(value) ? value.map(numberField).filter((item): item is number => item !== undefined) : [];
-}
-
 function stringField(value: unknown): string {
   return typeof value === "string" ? value : "";
 }

apps/desktop/src/components/mq/SendMessagePanel.vue

@@ -18,6 +18,7 @@ interface Props {
   mqSystemKind?: MqSystemKind;
   isFlatMqCluster?: boolean;
   supportsPeekMessages?: boolean;
+  fixedTopic?: boolean;
   /** Hide outer toolbar when embedded in a dialog. */
   embedded?: boolean;
 }
@@ -93,6 +94,11 @@ async function guardWritable() {
 }
 
 async function loadTopics() {
+  if (props.fixedTopic) {
+    availableTopics.value = props.topic ? [props.topic] : [];
+    topicName.value = props.topic?.shortName ?? "";
+    return;
+  }
   if (isRocketMqCluster.value) {
     await rocketMqTopicSelectRef.value?.loadTopics();
     return;
@@ -227,6 +233,9 @@ watch(
 watch(
   () => props.topic,
   (newTopic) => {
+    if (props.fixedTopic) {
+      availableTopics.value = newTopic ? [newTopic] : [];
+    }
     if (newTopic) {
       topicName.value = newTopic.shortName;
     }
@@ -255,7 +264,8 @@ watch(
 
       <div class="form-group">
         <label>{{ t("mqMessages.targetTopic") }} <span class="required">*</span></label>
-        <RocketMqTopicSelect v-if="isRocketMqCluster" ref="rocketMqTopicSelectRef" v-model="topicName" :connection-id="connectionId" :tenant="tenant" :namespace="namespace" grouping="business" :show-type-filter="false" :disabled="readOnly" @loaded="handleRocketMqTopicsLoaded" />
+        <input v-if="fixedTopic" :value="topicName" type="text" class="topic-input" disabled />
+        <RocketMqTopicSelect v-else-if="isRocketMqCluster" ref="rocketMqTopicSelectRef" v-model="topicName" :connection-id="connectionId" :tenant="tenant" :namespace="namespace" grouping="business" :show-type-filter="false" :disabled="readOnly" @loaded="handleRocketMqTopicsLoaded" />
         <template v-else>
           <div class="topic-select-row">
             <input v-model="topicName" :list="topicListId" :disabled="readOnly || topicsLoading" class="topic-input" :placeholder="topicsLoading ? t('mqMessages.topicLoading') : t('mqMessages.topicSearchPlaceholder')" autocomplete="off" />

apps/desktop/src/components/mq/__tests__/KafkaMessagesPanel.spec.ts

@@ -0,0 +1,155 @@
+// @vitest-environment happy-dom
+
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { createApp, nextTick, type App } from "vue";
+
+const backend = vi.hoisted(() => ({
+  mqListTopics: vi.fn(),
+  mqGetTopicStats: vi.fn(),
+  mqPeekMessages: vi.fn(),
+  mqSendMessage: vi.fn(),
+}));
+
+vi.mock("vue-i18n", () => ({
+  useI18n: () => ({ t: (key: string) => key }),
+}));
+vi.mock("@/lib/backend/api", () => backend);
+vi.mock("@/composables/useMqMutationGuard", () => ({
+  useMqMutationGuard: () => ({ confirmMqWrite: vi.fn().mockResolvedValue(true) }),
+}));
+vi.mock("@/components/ui/select", async () => (await import("./selectStub")).createSelectStub());
+vi.mock("@lucide/vue", async () => {
+  const { defineComponent, h } = await import("vue");
+  const Icon = defineComponent({ name: "Icon", setup: () => () => h("i") });
+  return { Copy: Icon, Search: Icon };
+});
+
+import KafkaMessagesPanel from "@/components/mq/KafkaMessagesPanel.vue";
+
+const TOPIC = { name: "events", shortName: "events", persistent: true, partitioned: true, partitions: 2 };
+const STATS = {
+  msgRateIn: 0,
+  msgRateOut: 0,
+  msgThroughputIn: 0,
+  msgThroughputOut: 0,
+  storageSize: 0,
+  backlogSize: 0,
+  msgInCounter: 12,
+  msgOutCounter: 0,
+  subscriptionCount: 0,
+  producerCount: 0,
+  raw: {
+    partitionStats: [
+      { partition: 1, beginOffset: 4, endOffset: 10, messageCount: 6 },
+      { partition: 0, beginOffset: 2, endOffset: 8, messageCount: 6 },
+    ],
+  },
+};
+
+let app: App<Element> | null = null;
+let root: HTMLDivElement | null = null;
+
+async function flushUi() {
+  await Promise.resolve();
+  await Promise.resolve();
+  await nextTick();
+}
+
+async function mountPanel(onTopicSelected = vi.fn()) {
+  root = document.createElement("div");
+  document.body.appendChild(root);
+  app = createApp(KafkaMessagesPanel, {
+    connectionId: "mq-1",
+    tenant: "_flat_mq",
+    namespace: "_flat_mq",
+    topic: TOPIC,
+    canSendMessage: true,
+    onTopicSelected,
+  });
+  app.mount(root);
+  await flushUi();
+  return root;
+}
+
+async function selectTopic(panel: HTMLElement, name: string) {
+  const trigger = panel.querySelector<HTMLButtonElement>(".topic-combobox-trigger");
+  if (!trigger) throw new Error("Topic selector trigger not found");
+  trigger.dispatchEvent(new MouseEvent("mousedown", { bubbles: true }));
+  await flushUi();
+  const option = [...panel.querySelectorAll<HTMLButtonElement>(".topic-combobox-option")].find((button) => button.textContent?.trim() === name);
+  if (!option) throw new Error(`Topic option ${name} not found`);
+  option.dispatchEvent(new MouseEvent("mousedown", { bubbles: true }));
+  await flushUi();
+}
+
+beforeEach(() => {
+  backend.mqListTopics.mockReset().mockResolvedValue([TOPIC, { ...TOPIC, name: "payments", shortName: "payments" }]);
+  backend.mqGetTopicStats.mockReset().mockResolvedValue(STATS);
+  backend.mqPeekMessages.mockReset().mockResolvedValue([]);
+  backend.mqSendMessage.mockReset().mockResolvedValue({ topic: TOPIC.shortName, partition: 0, offset: 1 });
+});
+
+afterEach(() => {
+  app?.unmount();
+  app = null;
+  root?.remove();
+  root = null;
+});
+
+describe("KafkaMessagesPanel", () => {
+  it("keeps topic selection, partition offsets, and newest-message browsing together", async () => {
+    const panel = await mountPanel();
+    await expect.poll(() => backend.mqListTopics.mock.calls.length).toBe(1);
+    await expect.poll(() => backend.mqGetTopicStats.mock.calls.length).toBeGreaterThan(0);
+    await flushUi();
+
+    const overview = panel.querySelector('[data-testid="kafka-partition-overview"]');
+    expect(overview?.textContent).toContain("mqMonitoring.tableBeginOffset");
+    expect(overview?.textContent).toContain("mqMonitoring.tableLogEndOffset");
+    expect(overview?.textContent).toContain("10");
+    expect(panel.querySelector('[data-testid="message-browser"]')).not.toBeNull();
+    expect(panel.querySelector(".send-message-panel")).not.toBeNull();
+    expect(panel.querySelector<HTMLInputElement>(".send-message-panel .topic-input")?.value).toBe("events");
+    expect(backend.mqListTopics).toHaveBeenCalledTimes(1);
+
+    const loadButton = [...panel.querySelectorAll<HTMLButtonElement>("button")].find((button) => button.textContent?.includes("mqMessages.loadMessages"));
+    if (!loadButton) throw new Error("Load messages button not found");
+    loadButton.click();
+    await flushUi();
+
+    expect(backend.mqPeekMessages).toHaveBeenCalledWith("mq-1", expect.objectContaining({ topic: "events" }), "__dbx_kafka_viewer__", 20, { startPosition: "latest" });
+  });
+
+  it("keeps message browsing available when partition stats fail", async () => {
+    backend.mqGetTopicStats.mockRejectedValueOnce(new Error("stats unavailable"));
+    const panel = await mountPanel();
+    await expect.poll(() => panel.textContent).toContain("stats unavailable");
+
+    expect(panel.querySelector('[data-testid="message-browser"]')).not.toBeNull();
+  });
+
+  it("keeps the user's selected topic across list refreshes and refresh failures", async () => {
+    const onTopicSelected = vi.fn();
+    const panel = await mountPanel(onTopicSelected);
+    await expect.poll(() => backend.mqListTopics.mock.calls.length).toBe(1);
+    await selectTopic(panel, "payments");
+    await expect.poll(() => backend.mqGetTopicStats.mock.calls.some((call) => call[1]?.topic === "payments")).toBe(true);
+    await expect.poll(() => onTopicSelected.mock.calls.some((call) => call[0]?.shortName === "payments")).toBe(true);
+    expect(panel.querySelector<HTMLInputElement>(".send-message-panel .topic-input")?.value).toBe("payments");
+    expect(backend.mqListTopics).toHaveBeenCalledTimes(1);
+
+    const refreshButton = panel.querySelector<HTMLButtonElement>(".panel-toolbar .btn-secondary");
+    if (!refreshButton) throw new Error("Topic refresh button not found");
+    refreshButton.click();
+    await expect.poll(() => backend.mqListTopics.mock.calls.length).toBe(2);
+    expect(panel.querySelector(".committed-value")?.textContent).toBe("payments");
+
+    const warning = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    backend.mqListTopics.mockRejectedValueOnce(new Error("list unavailable"));
+    refreshButton.click();
+    await expect.poll(() => backend.mqListTopics.mock.calls.length).toBe(3);
+    expect(panel.querySelector(".committed-value")?.textContent).toBe("payments");
+    expect(panel.querySelector('[data-testid="message-browser"]')).not.toBeNull();
+    warning.mockRestore();
+  });
+});

apps/desktop/src/components/mq/__tests__/MonitoringPanel.spec.ts

@@ -81,6 +81,7 @@ const KAFKA_STATS = {
 
 let app: App<Element> | null = null;
 let root: HTMLDivElement | null = null;
+const navigateTab = vi.fn();
 
 async function flushUi() {
   await Promise.resolve();
@@ -97,6 +98,7 @@ async function mountPanel() {
     namespace: "default",
     topic: TOPIC,
     mqSystemKind: "kafka",
+    onNavigateTab: navigateTab,
   });
   app.mount(root);
   await flushUi();
@@ -107,6 +109,7 @@ beforeEach(() => {
   backend.mqGetTopicStats.mockReset();
   backend.mqGetBacklog.mockReset();
   backend.mqPeekMessages.mockReset();
+  navigateTab.mockReset();
   backend.mqGetTopicStats.mockResolvedValue(KAFKA_STATS);
   backend.mqPeekMessages.mockResolvedValue([]);
 });
@@ -128,7 +131,13 @@ describe("MonitoringPanel Kafka message browser", () => {
     expect(panel.querySelector("textarea")).toBeNull();
     expect(browser.classList.contains("message-browser")).toBe(true);
     expect(browser.classList.contains("is-monitoring")).toBe(true);
-    expect(browser.querySelector('[data-testid="kafka-peek-start-position"]')).not.toBeNull();
+    expect(browser.querySelector('[data-testid="kafka-peek-start-position"]')).toBeNull();
+    expect(browser.querySelector('[data-testid="kafka-peek-partition"]')).not.toBeNull();
+
+    const fullQueryButton = [...panel.querySelectorAll<HTMLButtonElement>("button")].find((button) => button.textContent?.includes("mqMessages.queryTitle"));
+    if (!fullQueryButton) throw new Error("Full Kafka query button not found");
+    fullQueryButton.click();
+    expect(navigateTab).toHaveBeenCalledWith({ tab: "messages", topic: TOPIC });
 
     loadButton.click();
     await flushUi();

apps/desktop/src/components/mq/__tests__/RocketMqTopicSelect.spec.ts

@@ -0,0 +1,75 @@
+// @vitest-environment happy-dom
+
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { createApp, defineComponent, h, nextTick, ref, type App, type ComponentPublicInstance } from "vue";
+
+const backend = vi.hoisted(() => ({
+  mqListTopics: vi.fn(),
+}));
+
+vi.mock("vue-i18n", () => ({
+  useI18n: () => ({ t: (key: string) => key }),
+}));
+vi.mock("@/lib/backend/api", () => backend);
+vi.mock("@lucide/vue", async () => {
+  const { defineComponent: defineIcon, h: render } = await import("vue");
+  const Icon = defineIcon({ name: "Icon", setup: () => () => render("i") });
+  return { Search: Icon };
+});
+
+import RocketMqTopicSelect from "@/components/mq/shared/RocketMqTopicSelect.vue";
+
+type TopicSelectExposed = ComponentPublicInstance & { loading: boolean };
+
+let app: App<Element> | null = null;
+let root: HTMLDivElement | null = null;
+
+async function flushUi() {
+  await Promise.resolve();
+  await Promise.resolve();
+  await nextTick();
+}
+
+beforeEach(() => {
+  backend.mqListTopics.mockReset();
+});
+
+afterEach(() => {
+  app?.unmount();
+  app = null;
+  root?.remove();
+  root = null;
+});
+
+describe("RocketMqTopicSelect request state", () => {
+  it("clears loading when the request context becomes incomplete", async () => {
+    backend.mqListTopics.mockReturnValue(new Promise(() => undefined));
+    const tenant = ref<string>();
+    const selector = ref<TopicSelectExposed>();
+    const Wrapper = defineComponent({
+      setup() {
+        return () =>
+          h(RocketMqTopicSelect, {
+            ref: selector,
+            modelValue: "",
+            connectionId: "mq-1",
+            tenant: tenant.value,
+            namespace: "default",
+          });
+      },
+    });
+
+    root = document.createElement("div");
+    document.body.appendChild(root);
+    tenant.value = "tenant-a";
+    app = createApp(Wrapper);
+    app.mount(root);
+    await flushUi();
+    expect(selector.value?.loading).toBe(true);
+
+    tenant.value = undefined;
+    await flushUi();
+
+    expect(selector.value?.loading).toBe(false);
+  });
+});

apps/desktop/src/components/mq/shared/RocketMqTopicSelect.vue

@@ -46,6 +46,7 @@ const dropdownOpen = ref(false);
 const highlightIndex = ref(-1);
 const rootRef = ref<HTMLElement>();
 const inputRef = ref<HTMLInputElement>();
+let loadRequestVersion = 0;
 
 const listboxId = computed(() => `mq-topic-listbox-${props.connectionId}-${props.tenant ?? ""}-${props.namespace ?? ""}`);
 
@@ -180,14 +181,16 @@ function onDocumentPointerDown(event: PointerEvent) {
 }
 
 async function loadTopics() {
+  const requestVersion = ++loadRequestVersion;
   if (!props.tenant || !props.namespace) {
     topics.value = [];
+    loading.value = false;
     emit("loaded", []);
     return;
   }
   loading.value = true;
   try {
-    topics.value = await mqListTopics(
+    const loaded = await mqListTopics(
       props.connectionId,
       {
         tenant: props.tenant,
@@ -195,13 +198,13 @@ async function loadTopics() {
       },
       { includeNonPersistent: false },
     );
-    emit("loaded", topics.value);
+    if (requestVersion !== loadRequestVersion) return;
+    topics.value = loaded;
+    emit("loaded", loaded);
   } catch (e: unknown) {
     console.warn("[DBX] Failed to load RocketMQ topics:", e);
-    topics.value = [];
-    emit("loaded", []);
   } finally {
-    loading.value = false;
+    if (requestVersion === loadRequestVersion) loading.value = false;
   }
 }
 
@@ -227,7 +230,12 @@ watch(dropdownOpen, (open) => {
 
 watch(
   () => [props.tenant, props.namespace, props.connectionId],
-  () => {
+  (_context, previousContext) => {
+    if (previousContext) {
+      loadRequestVersion += 1;
+      topics.value = [];
+      emit("loaded", []);
+    }
     void loadTopics();
   },
   { immediate: true },

apps/desktop/src/i18n/locales/en.ts

@@ -6779,7 +6779,7 @@ export default {
     kafkaLatest: "Recent N messages (default)",
     kafkaEarliest: "Earliest messages",
     kafkaOffset: "From offset",
-    kafkaLatestHint: "Reads a balanced window near the end of each partition and merges by timestamp newest first.",
+    kafkaLatestHint: "Reads up to N candidates from each partition, then returns the newest N by timestamp across the topic.",
     kafkaEarliestHint: "Reads history forward from each selected partition.",
     kafkaOffsetHint: "Reads forward from the specified offset in the chosen partition, or every partition when left empty.",
     messagesLoading: "Loading messages...",

apps/desktop/src/i18n/locales/es.ts

@@ -6719,7 +6719,7 @@ export default withEnglishFallback({
     kafkaLatest: "Últimos N mensajes (predeterminado)",
     kafkaEarliest: "Mensajes más antiguos",
     kafkaOffset: "Desde offset",
-    kafkaLatestHint: "Lee una ventana equilibrada cerca del final de cada partición y combina por marca de tiempo con los más nuevos primero.",
+    kafkaLatestHint: "Lee hasta N candidatos por partición y devuelve los N más recientes del tema por marca de tiempo.",
     kafkaEarliestHint: "Lee el historial hacia adelante desde cada partición seleccionada.",
     kafkaOffsetHint: "Lee hacia adelante desde el offset indicado en la partición elegida, o en todas si se deja vacía.",
     messagesLoading: "Cargando mensajes...",

apps/desktop/src/i18n/locales/it.ts

@@ -6719,7 +6719,7 @@ export default withEnglishFallback({
     kafkaLatest: "Ultimi N messaggi (predefinito)",
     kafkaEarliest: "Primi messaggi",
     kafkaOffset: "Da offset",
-    kafkaLatestHint: "Legge una finestra bilanciata vicino alla fine di ogni partizione e unisce per timestamp con i più recenti prima.",
+    kafkaLatestHint: "Legge fino a N candidati per partizione e restituisce i N messaggi più recenti del topic in base al timestamp.",
     kafkaEarliestHint: "Legge la cronologia in avanti da ogni partizione selezionata.",
     kafkaOffsetHint: "Legge in avanti dall'offset indicato nella partizione scelta, o in tutte se lasciata vuota.",
     messagesLoading: "Caricamento messaggi...",

apps/desktop/src/i18n/locales/ja.ts

@@ -6772,7 +6772,7 @@ export default withEnglishFallback({
     kafkaLatest: "最新 N 件（デフォルト）",
     kafkaEarliest: "最も古いメッセージ",
     kafkaOffset: "Offset から読み取る",
-    kafkaLatestHint: "各パーティションの末尾付近から均等に読み取り、タイムスタンプの新しい順に統合します。",
+    kafkaLatestHint: "各パーティションから最大 N 件の候補を読み取り、トピック全体でタイムスタンプが新しい N 件を返します。",
     kafkaEarliestHint: "選択した各パーティションの先頭から昇順に読み取ります。",
     kafkaOffsetHint: "指定 Offset から、選択したパーティション（空欄ならすべて）を昇順に読み取ります。",
     messagesLoading: "メッセージを読み込み中...",

apps/desktop/src/i18n/locales/ko.ts

@@ -6255,7 +6255,7 @@ export default withEnglishFallback({
     kafkaLatest: "최근 N개 메시지(기본값)",
     kafkaEarliest: "가장 이른 메시지",
     kafkaOffset: "오프셋부터 읽기",
-    kafkaLatestHint: "각 파티션 끝 부근에서 균등하게 읽고 타임스탬프 내림차순으로 병합합니다.",
+    kafkaLatestHint: "파티션마다 최대 N개의 후보를 읽은 뒤 토픽 전체에서 타임스탬프가 가장 최신인 N개를 반환합니다.",
     kafkaEarliestHint: "선택한 각 파티션의 처음부터 오름차순으로 읽습니다.",
     kafkaOffsetHint: "지정한 오프셋부터 선택한 파티션(비워 두면 전체)을 오름차순으로 읽습니다.",
     messagesLoading: "메시지를 불러오는 중...",

apps/desktop/src/i18n/locales/pt-BR.ts

@@ -6721,7 +6721,7 @@ export default withEnglishFallback({
     kafkaLatest: "Últimas N mensagens (padrão)",
     kafkaEarliest: "Mensagens mais antigas",
     kafkaOffset: "A partir do offset",
-    kafkaLatestHint: "Lê uma janela equilibrada perto do fim de cada partição e mescla por data e hora com as mais novas primeiro.",
+    kafkaLatestHint: "Lê até N candidatos por partição e retorna as N mensagens mais recentes do tópico por data e hora.",
     kafkaEarliestHint: "Lê o histórico em ordem crescente de cada partição selecionada.",
     kafkaOffsetHint: "Lê em ordem crescente a partir do offset informado na partição escolhida, ou em todas se deixada vazia.",
     messagesLoading: "Carregando mensagens...",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -6766,7 +6766,7 @@ export default withEnglishFallback({
     kafkaLatest: "最近 N 条（默认）",
     kafkaEarliest: "最早消息",
     kafkaOffset: "从指定 Offset 读取",
-    kafkaLatestHint: "从各分区末尾附近均衡读取，按时间倒序合并。",
+    kafkaLatestHint: "每个分区最多读取 N 条候选消息，再按时间全局倒序返回最新 N 条。",
     kafkaEarliestHint: "从每个选定分区的最早可读消息开始正序读取。",
     kafkaOffsetHint: "从指定 Offset 开始正序读取；填写分区时仅读该分区，留空则对全部分区分别读取。",
     messagesLoading: "消息加载中...",

apps/desktop/src/i18n/locales/zh-TW.ts

@@ -6712,7 +6712,7 @@ export default withEnglishFallback({
     kafkaLatest: "最近 N 則（預設）",
     kafkaEarliest: "最早訊息",
     kafkaOffset: "從指定 Offset 讀取",
-    kafkaLatestHint: "從各分區末尾附近平均讀取，依時間倒序合併。",
+    kafkaLatestHint: "每個分區最多讀取 N 筆候選訊息，再依時間全域倒序回傳最新 N 筆。",
     kafkaEarliestHint: "從每個選定分區最早可讀的訊息開始正序讀取。",
     kafkaOffsetHint: "從指定 Offset 開始正序讀取；填寫分區時只讀該分區，留空則對全部分區分別讀取。",
     messagesLoading: "訊息載入中...",

apps/desktop/src/lib/__tests__/mq/kafkaTopicStats.spec.ts

@@ -0,0 +1,22 @@
+import { describe, expect, it } from "vitest";
+import { extractKafkaPartitionRows, isKafkaStatsPayload } from "@/lib/mq/kafkaTopicStats";
+
+describe("Kafka topic stats", () => {
+  it("normalizes and sorts partition rows", () => {
+    expect(
+      extractKafkaPartitionRows({
+        partitionStats: [{ partition: 2, beginOffset: 8, endOffset: 20, messageCount: 12, leader: 3, replicas: [3, 4], isr: [3] }, { partition: 0, beginOffset: 1, endOffset: 4, messageCount: 3 }, null],
+      }),
+    ).toEqual([
+      { partition: 0, beginOffset: 1, endOffset: 4, messageCount: 3, leader: -1, replicas: [], isr: [] },
+      { partition: 2, beginOffset: 8, endOffset: 20, messageCount: 12, leader: 3, replicas: [3, 4], isr: [3] },
+    ]);
+  });
+
+  it("recognizes Kafka summaries and rejects unrelated payloads", () => {
+    expect(isKafkaStatsPayload({ partitionStats: [] })).toBe(true);
+    expect(isKafkaStatsPayload({ partitions: 2, replicationFactor: 1, totalMessages: 4 })).toBe(true);
+    expect(isKafkaStatsPayload({ partitions: 2 })).toBe(false);
+    expect(extractKafkaPartitionRows(undefined)).toEqual([]);
+  });
+});

apps/desktop/src/lib/mq/__tests__/mqConsoleDefaults.spec.ts

@@ -77,6 +77,13 @@ describe("mqConsoleDefaults", () => {
     expect(defaultMqCapabilitiesForSystemKind("pulsar").supportsClientConnections).toBeFalsy();
   });
 
+  it("exposes Kafka message browsing through peek capability without query capability", () => {
+    const caps = { ...defaultMqCapabilitiesForSystemKind("kafka"), supportsSendMessage: false };
+    expect(caps.supportsPeekMessages).toBe(true);
+    expect(caps.supportsMessageQuery).toBe(false);
+    expect(resolveAvailableMqTabs({ systemKind: "kafka", capabilities: caps })).toContain("messages");
+  });
+
   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"]);

apps/desktop/src/lib/mq/kafkaTopicStats.ts

@@ -0,0 +1,49 @@
+export interface KafkaPartitionStatsRow {
+  partition: number;
+  beginOffset: number;
+  endOffset: number;
+  messageCount: number;
+  leader: number;
+  replicas: number[];
+  isr: number[];
+}
+
+export function extractKafkaPartitionRows(raw: unknown): KafkaPartitionStatsRow[] {
+  return arrayObjects(objectRecord(raw).partitionStats)
+    .map((body) => ({
+      partition: numberField(body.partition) ?? 0,
+      beginOffset: numberField(body.beginOffset) ?? 0,
+      endOffset: numberField(body.endOffset) ?? 0,
+      messageCount: numberField(body.messageCount) ?? 0,
+      leader: numberField(body.leader) ?? -1,
+      replicas: numberArrayField(body.replicas),
+      isr: numberArrayField(body.isr),
+    }))
+    .sort((left, right) => left.partition - right.partition);
+}
+
+export function isKafkaStatsPayload(raw: unknown): boolean {
+  const root = objectRecord(raw);
+  return Array.isArray(root.partitionStats) || (numberField(root.partitions) !== undefined && numberField(root.replicationFactor) !== undefined && numberField(root.totalMessages) !== undefined);
+}
+
+function objectRecord(value: unknown): Record<string, unknown> {
+  return value && typeof value === "object" && !Array.isArray(value) ? (value as Record<string, unknown>) : {};
+}
+
+function arrayObjects(value: unknown): Record<string, unknown>[] {
+  return Array.isArray(value) ? value.filter((item): item is Record<string, unknown> => !!item && typeof item === "object" && !Array.isArray(item)) : [];
+}
+
+function numberField(value: unknown): number | undefined {
+  if (typeof value === "number" && Number.isFinite(value)) return value;
+  if (typeof value === "string" && value.trim()) {
+    const parsed = Number(value);
+    return Number.isFinite(parsed) ? parsed : undefined;
+  }
+  return undefined;
+}
+
+function numberArrayField(value: unknown): number[] {
+  return Array.isArray(value) ? value.map(numberField).filter((item): item is number => item !== undefined) : [];
+}

apps/desktop/src/lib/mq/mqConsoleDefaults.ts

@@ -177,7 +177,7 @@ export function resolveAvailableMqTabs(options: { systemKind?: MqSystemKind; cap
   if (capabilities.supportsSubscriptions) tabs.push("subscriptions");
   tabs.push("monitoring");
   tabs.push("clients");
-  if (capabilities.supportsSendMessage) tabs.push("messages");
+  if (capabilities.supportsSendMessage || (systemKind === "kafka" && capabilities.supportsPeekMessages)) tabs.push("messages");
   tabs.push("broker");
   // RabbitMQ lights this tab via virtual-host policies instead of Pulsar rates/quotas.
   if (capabilities.supportsRateLimits || capabilities.supportsBacklogQuota || capabilities.supportsRetention || capabilities.supportsPolicies) {