feat(rocketmq): improve connections, topics, and message queries

.gitattributes

@@ -1,6 +1,7 @@
 # Keep formatter-sensitive source files consistent across platforms.
 apps/desktop/src/**/*.ts text eol=lf
 apps/desktop/src/**/*.vue text eol=lf
+src-tauri/windows/nsis/**/*.nsi text eol=lf
 
 # Keep tests available for review and CI without counting them as shipped code
 # in GitHub's repository language breakdown.

agents/drivers/rocketmq/src/main/java/com/dbx/agent/rocketmq/RocketMqAgent.java

@@ -61,6 +61,9 @@ public final class RocketMqAgent {
     private static final int DEFAULT_LIST_LIMIT = 200;
     private static final int CONSUMER_GROUP_ENRICH_CONCURRENCY = 8;
     private static final long CONSUMER_GROUP_ENRICH_BUDGET_MS = 8_000;
+    /** Nameserver route enrichment for listTopics fallback when bulk broker config is unavailable. */
+    private static final int TOPIC_LIST_ROUTE_CONCURRENCY = 8;
+    private static final long TOPIC_LIST_ROUTE_BUDGET_MS = 8_000;
     private static final String AUTO_CREATE_TOPIC_KEY = "TBW102";
     /** RocketMQ 5.x topic attribute key for message type (NORMAL/DELAY/FIFO/TRANSACTION). */
     private static final String TOPIC_MESSAGE_TYPE_ATTRIBUTE = "message.type";
@@ -364,47 +367,35 @@ public final class RocketMqAgent {
     private static Object connect(JsonObject params) throws Exception {
         JsonObject conn = connectionObject(params);
         DefaultMQAdminExt nextAdmin = null;
-        DefaultMQProducer nextProducer = null;
         try {
             nextAdmin = buildAdminClient(conn);
-            nextAdmin.examineBrokerClusterInfo();
-            nextProducer = buildProducer(conn);
+            // One ClusterInfo per connect — reuse for name/broker resolution and test payload.
+            ClusterInfo clusterInfo = nextAdmin.examineBrokerClusterInfo();
             closeClients();
             adminClient = nextAdmin;
-            producer = nextProducer;
             cachedConnection = conn.deepCopy();
-            cachedClusterName = resolveClusterName(nextAdmin, conn);
-            cachedBrokerAddr = resolveBrokerAddr(nextAdmin, conn);
-            return Collections.singletonMap("ok", true);
+            cachedClusterName = resolveClusterName(clusterInfo, conn);
+            cachedBrokerAddr = resolveBrokerAddr(nextAdmin, conn, clusterInfo);
+            return buildClusterTestResult(clusterInfo, nextAdmin, conn);
         } catch (Exception e) {
             if (nextAdmin != null) {
                 nextAdmin.shutdown();
             }
-            if (nextProducer != null) {
-                nextProducer.shutdown();
-            }
             throw e;
         }
     }
 
     private static Object testConnection(JsonObject params) throws Exception {
         JsonObject conn = connectionObject(params);
+        if (adminClient != null && cachedConnection != null && connectionMatches(cachedConnection, conn)) {
+            ClusterInfo clusterInfo = adminClient.examineBrokerClusterInfo();
+            return buildClusterTestResult(clusterInfo, adminClient, conn);
+        }
         DefaultMQAdminExt probe = null;
         try {
             probe = buildAdminClient(conn);
             ClusterInfo clusterInfo = probe.examineBrokerClusterInfo();
-            String clusterName = resolveClusterName(clusterInfo, conn);
-            List<Map<String, Object>> brokers = brokerNodes(clusterInfo);
-            boolean aclEnabled = probeAclSupport(probe);
-
-            Map<String, Object> result = new LinkedHashMap<>();
-            result.put("ok", true);
-            result.put("clusterId", clusterName);
-            result.put("brokers", brokers);
-            result.put("nodeCount", brokers.size());
-            result.put("controller", brokers.isEmpty() ? null : brokers.get(0));
-            result.put("aclEnabled", aclEnabled);
-            return result;
+            return buildClusterTestResult(clusterInfo, probe, conn);
         } finally {
             if (probe != null) {
                 probe.shutdown();
@@ -412,6 +403,33 @@ public final class RocketMqAgent {
         }
     }
 
+    private static Map<String, Object> buildClusterTestResult(
+        ClusterInfo clusterInfo,
+        DefaultMQAdminExt admin,
+        JsonObject conn
+    ) throws Exception {
+        String clusterName = resolveClusterName(clusterInfo, conn);
+        List<Map<String, Object>> brokers = brokerNodes(clusterInfo);
+        boolean aclEnabled = probeAclSupport(admin, clusterInfo);
+
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("ok", true);
+        result.put("clusterId", clusterName);
+        result.put("brokers", brokers);
+        result.put("nodeCount", brokers.size());
+        result.put("controller", brokers.isEmpty() ? null : brokers.get(0));
+        result.put("aclEnabled", aclEnabled);
+        return result;
+    }
+
+    static boolean connectionMatches(JsonObject cached, JsonObject requested) {
+        return namesrvAddr(cached).equals(namesrvAddr(requested))
+            && clusterName(cached).equals(clusterName(requested))
+            && brokerAddress(cached).equals(brokerAddress(requested))
+            && credential(cached, "access_key", "accessKey").equals(credential(requested, "access_key", "accessKey"))
+            && credential(cached, "secret_key", "secretKey").equals(credential(requested, "secret_key", "secretKey"));
+    }
+
     private static void closeClients() {
         if (adminClient != null) {
             adminClient.shutdown();
@@ -430,47 +448,129 @@ public final class RocketMqAgent {
         DefaultMQAdminExt admin = requireAdmin();
         String keyword = stringOrEmpty(params, "keyword").toLowerCase(Locale.ROOT);
         int offset = Math.max(0, intOrDefault(params, "offset", 0));
+        // limit <= 0: return all rows from offset (legacy single-shot signal).
+        // Large positive limits (e.g. Integer.MAX_VALUE) also return the full remaining
+        // catalog and stay compatible with older agents that coerced limit<=0 to 200.
         int limit = intOrDefault(params, "limit", DEFAULT_LIST_LIMIT);
-        if (limit <= 0) {
-            limit = DEFAULT_LIST_LIMIT;
+        JsonObject conn = connectionObject(params);
+
+        // One ClusterInfo per list call — avoid repeated examineBrokerClusterInfo in helpers.
+        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
+        Set<String> brokerSystemTopics = collectBrokerSystemTopics(admin);
+        Set<String> brokerNames = collectBrokerNames(clusterInfo);
+        String cluster = resolveClusterName(clusterInfo, conn);
+        BrokerTopicConfigSnapshot brokerSnapshot = collectBrokerTopicConfigSnapshot(admin, conn, clusterInfo);
+        Map<String, TopicConfig> brokerTopics = brokerSnapshot.topics();
+        Map<String, Map<String, String>> topicAttributes = topicAttributesFromConfigs(brokerTopics);
+
+        Set<String> nameserverTopics = Set.of();
+        if (!brokerSnapshot.complete() || brokerTopics.isEmpty()) {
+            // Reuse resolved cluster — do not call examineBrokerClusterInfo again via clusterName(conn, admin).
+            TopicList topicList = fetchTopicList(admin, cluster);
+            nameserverTopics = topicList.getTopicList();
+        }
+        Set<String> topicNames = topicCatalogNames(brokerSnapshot, nameserverTopics);
+
+        List<Map<String, Object>> topics = buildTopicCatalogRows(
+            topicNames,
+            brokerTopics,
+            topicAttributes,
+            brokerSystemTopics,
+            brokerNames,
+            cluster,
+            keyword,
+            admin::examineTopicRouteInfo,
+            TOPIC_LIST_ROUTE_CONCURRENCY,
+            TOPIC_LIST_ROUTE_BUDGET_MS
+        );
+        topics.sort(Comparator.comparing(m -> String.valueOf(m.get("name"))));
+
+        int total = topics.size();
+        List<Map<String, Object>> page = paginate(topics, offset, limit);
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("topics", page);
+        result.put("total", total);
+        result.put("offset", offset);
+        result.put("limit", limit);
+        return result;
+    }
+
+    @FunctionalInterface
+    interface TopicRouteLookup {
+        TopicRouteData load(String topic) throws Exception;
+    }
+
+    @FunctionalInterface
+    interface BrokerTopicConfigLookup {
+        TopicConfigSerializeWrapper load(String brokerAddr) throws Exception;
+    }
+
+    static final class BrokerTopicConfigSnapshot {
+        private final Map<String, TopicConfig> topics;
+        private final boolean complete;
+
+        private BrokerTopicConfigSnapshot(Map<String, TopicConfig> topics, boolean complete) {
+            this.topics = topics;
+            this.complete = complete;
         }
 
-        TopicList topicList = fetchTopicList(admin, connectionObject(params));
-        Set<String> brokerSystemTopics = collectBrokerSystemTopics(admin);
-        Set<String> brokerNames = collectBrokerNames(admin);
-        String cluster = clusterName(params, admin);
-        JsonObject conn = connectionObject(params);
-        Map<String, TopicConfig> brokerTopics = collectBrokerTopicConfigs(admin, conn);
-        Map<String, Map<String, String>> topicAttributes = topicAttributesFromConfigs(brokerTopics);
-        // Prefer broker topic configs (Dashboard ground truth); nameserver routes can outlive broker deletion.
-        Set<String> topicNames = brokerTopics.isEmpty()
-            ? new TreeSet<>(topicList.getTopicList())
-            : brokerTopics.keySet();
+        Map<String, TopicConfig> topics() {
+            return topics;
+        }
+
+        boolean complete() {
+            return complete;
+        }
+    }
+
+    static Set<String> topicCatalogNames(
+        BrokerTopicConfigSnapshot brokerSnapshot, Set<String> nameserverTopics) {
+        if (brokerSnapshot.complete() && !brokerSnapshot.topics().isEmpty()) {
+            return new TreeSet<>(brokerSnapshot.topics().keySet());
+        }
+        Set<String> topics = new TreeSet<>(brokerSnapshot.topics().keySet());
+        topics.addAll(nameserverTopics);
+        return topics;
+    }
+
+    /**
+     * Build topic catalog rows from names + optional bulk broker configs.
+     * When bulk configs are empty, partition counts are filled via a single budgeted route RPC
+     * per topic; per-topic examineTopicConfig / resolveTopicMessageType is never used (that path
+     * hits the same broker that already failed getAllTopicConfig and can stall for minutes).
+     */
+    static List<Map<String, Object>> buildTopicCatalogRows(
+        Set<String> topicNames,
+        Map<String, TopicConfig> brokerTopics,
+        Map<String, Map<String, String>> topicAttributes,
+        Set<String> brokerSystemTopics,
+        Set<String> brokerNames,
+        String cluster,
+        String keyword,
+        TopicRouteLookup routeLookup,
+        int routeConcurrency,
+        long routeBudgetMs
+    ) {
+        String keywordLower = keyword == null ? "" : keyword.toLowerCase(Locale.ROOT);
+        boolean hasBrokerCatalog = brokerTopics != null && !brokerTopics.isEmpty();
         List<Map<String, Object>> topics = new ArrayList<>();
+        List<Map<String, Object>> needsRoute = new ArrayList<>();
         for (String topic : topicNames) {
-            if (!keyword.isBlank() && !topic.toLowerCase(Locale.ROOT).contains(keyword)) {
+            if (!keywordLower.isBlank() && !topic.toLowerCase(Locale.ROOT).contains(keywordLower)) {
                 continue;
             }
-            TopicConfig brokerConfig = brokerTopics.get(topic);
-            int partitions = brokerConfig == null ? 1 : Math.max(brokerConfig.getReadQueueNums(), 1);
-            if (brokerConfig == null) {
-                try {
-                    TopicRouteData route = admin.examineTopicRouteInfo(topic);
-                    if (route.getQueueDatas() != null && !route.getQueueDatas().isEmpty()) {
-                        partitions = Math.max(route.getQueueDatas().get(0).getReadQueueNums(), 1);
-                    }
-                } catch (Exception ignored) {
-                    // Stale nameserver-only topics are skipped below when broker configs are available.
-                    if (!brokerTopics.isEmpty()) {
-                        continue;
-                    }
-                }
+            TopicConfig brokerConfig = hasBrokerCatalog ? brokerTopics.get(topic) : null;
+            // Nameserver-only stale topics are skipped when broker catalog is the source of truth.
+            if (hasBrokerCatalog && brokerConfig == null) {
+                continue;
             }
-            Map<String, String> attributes = topicAttributes.get(topic);
-            if (isUserTopic(topic) && readTopicMessageTypeAttribute(attributes) == null) {
-                String resolved = brokerConfig == null
-                    ? resolveTopicMessageType(admin, conn, topic)
-                    : readTopicMessageType(brokerConfig);
+            int partitions = brokerConfig == null ? 1 : Math.max(brokerConfig.getReadQueueNums(), 1);
+            Map<String, String> attributes = topicAttributes == null ? null : topicAttributes.get(topic);
+            // Type from bulk config attributes only — never examineTopicConfig per topic on fallback.
+            if (brokerConfig != null
+                && isUserTopic(topic)
+                && readTopicMessageTypeAttribute(attributes) == null) {
+                String resolved = readTopicMessageType(brokerConfig);
                 if (resolved != null && !resolved.isBlank()) {
                     attributes = attributes == null ? new HashMap<>() : new HashMap<>(attributes);
                     attributes.put("+" + TOPIC_MESSAGE_TYPE_ATTRIBUTE, resolved);
@@ -488,17 +588,71 @@ public final class RocketMqAgent {
             row.put("internal", internal);
             row.put("messageType", messageType);
             topics.add(row);
+            if (brokerConfig == null) {
+                needsRoute.add(row);
+            }
         }
-        topics.sort(Comparator.comparing(m -> String.valueOf(m.get("name"))));
+        if (!needsRoute.isEmpty() && routeLookup != null) {
+            enrichTopicPartitions(needsRoute, routeLookup, routeConcurrency, routeBudgetMs);
+        }
+        return topics;
+    }
 
-        int total = topics.size();
-        List<Map<String, Object>> page = paginate(topics, offset, limit);
-        Map<String, Object> result = new LinkedHashMap<>();
-        result.put("topics", page);
-        result.put("total", total);
-        result.put("offset", offset);
-        result.put("limit", limit);
-        return result;
+    static int partitionsFromRoute(TopicRouteData route) {
+        if (route == null || route.getQueueDatas() == null || route.getQueueDatas().isEmpty()) {
+            return 1;
+        }
+        return Math.max(route.getQueueDatas().get(0).getReadQueueNums(), 1);
+    }
+
+    /**
+     * Fill {@code partitions} via nameserver route lookups with bounded concurrency and a global
+     * time budget. Unfinished/failed topics keep the default partitions=1.
+     */
+    static void enrichTopicPartitions(
+        List<Map<String, Object>> rows,
+        TopicRouteLookup lookup,
+        int concurrency,
+        long budgetMs) {
+        if (rows.isEmpty() || lookup == null) {
+            return;
+        }
+        int workers = Math.max(1, Math.min(concurrency, rows.size()));
+        ExecutorService executor = Executors.newFixedThreadPool(workers, runnable -> {
+            Thread thread = new Thread(runnable, "dbx-rocketmq-topic-route");
+            thread.setDaemon(true);
+            return thread;
+        });
+        List<Callable<Integer>> tasks = new ArrayList<>(rows.size());
+        for (Map<String, Object> row : rows) {
+            String topic = String.valueOf(row.get("name"));
+            tasks.add(() -> {
+                try {
+                    return partitionsFromRoute(lookup.load(topic));
+                } catch (Exception ignored) {
+                    return 1;
+                }
+            });
+        }
+        List<Future<Integer>> results = Collections.emptyList();
+        try {
+            results = executor.invokeAll(tasks, Math.max(1, budgetMs), TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } finally {
+            executor.shutdownNow();
+        }
+        for (int index = 0; index < rows.size(); index++) {
+            Map<String, Object> row = rows.get(index);
+            if (index < results.size() && !results.get(index).isCancelled()) {
+                try {
+                    row.put("partitions", results.get(index).get());
+                } catch (Exception ignored) {
+                    // Keep default partitions=1 when enrichment fails.
+                }
+            }
+            row.putIfAbsent("partitions", 1);
+        }
     }
 
     private static Object createTopic(JsonObject params) throws Exception {
@@ -1819,9 +1973,8 @@ public final class RocketMqAgent {
         return addr;
     }
 
-    private static TopicList fetchTopicList(DefaultMQAdminExt admin, JsonObject conn) throws Exception {
-        String cluster = clusterName(conn, admin);
-        if (!cluster.isBlank()) {
+    private static TopicList fetchTopicList(DefaultMQAdminExt admin, String cluster) throws Exception {
+        if (cluster != null && !cluster.isBlank()) {
             return admin.fetchTopicsByCLuster(cluster);
         }
         return admin.fetchAllTopicList();
@@ -1865,32 +2018,45 @@ public final class RocketMqAgent {
 
     static List<String> resolveMasterBrokerAddrs(
         DefaultMQAdminExt admin, JsonObject conn, String brokerNameFilter) throws Exception {
-        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
-        LinkedHashSet<String> addrs = new LinkedHashSet<>();
-        if (clusterInfo.getBrokerAddrTable() != null) {
-            for (BrokerData broker : clusterInfo.getBrokerAddrTable().values()) {
-                if (brokerNameFilter != null && !brokerNameFilter.isBlank()
-                    && !brokerNameFilter.equals(broker.getBrokerName())) {
-                    continue;
-                }
-                String masterAddr = null;
-                if (broker.getBrokerAddrs() != null && broker.getBrokerAddrs().containsKey(0L)) {
-                    masterAddr = broker.getBrokerAddrs().get(0L);
-                }
-                if (masterAddr == null || masterAddr.isBlank()) {
-                    masterAddr = broker.selectBrokerAddr();
-                }
-                if (masterAddr != null && !masterAddr.isBlank()) {
-                    addrs.add(remapBrokerAddrForClient(masterAddr, conn));
-                }
-            }
-        }
+        return resolveMasterBrokerAddrs(admin, conn, brokerNameFilter, admin.examineBrokerClusterInfo());
+    }
+
+    static List<String> resolveMasterBrokerAddrs(
+        DefaultMQAdminExt admin, JsonObject conn, String brokerNameFilter, ClusterInfo clusterInfo)
+        throws Exception {
+        LinkedHashSet<String> addrs = masterBrokerAddrsFromClusterInfo(clusterInfo, conn, brokerNameFilter);
         if (addrs.isEmpty()) {
-            addrs.add(resolveBrokerAddr(admin, conn));
+            addrs.add(resolveBrokerAddr(admin, conn, clusterInfo));
         }
         return new ArrayList<>(addrs);
     }
 
+    /** Extract remapped master broker addresses from a ClusterInfo snapshot. */
+    static LinkedHashSet<String> masterBrokerAddrsFromClusterInfo(
+        ClusterInfo clusterInfo, JsonObject conn, String brokerNameFilter) {
+        LinkedHashSet<String> addrs = new LinkedHashSet<>();
+        if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+            return addrs;
+        }
+        for (BrokerData broker : clusterInfo.getBrokerAddrTable().values()) {
+            if (brokerNameFilter != null && !brokerNameFilter.isBlank()
+                && !brokerNameFilter.equals(broker.getBrokerName())) {
+                continue;
+            }
+            String masterAddr = null;
+            if (broker.getBrokerAddrs() != null && broker.getBrokerAddrs().containsKey(0L)) {
+                masterAddr = broker.getBrokerAddrs().get(0L);
+            }
+            if (masterAddr == null || masterAddr.isBlank()) {
+                masterAddr = broker.selectBrokerAddr();
+            }
+            if (masterAddr != null && !masterAddr.isBlank()) {
+                addrs.add(remapBrokerAddrForClient(masterAddr, conn));
+            }
+        }
+        return addrs;
+    }
+
     private static void applyTopicConfigValue(TopicConfig config, String key, String value) {
         if (value == null) {
             return;
@@ -2071,9 +2237,11 @@ public final class RocketMqAgent {
         };
     }
 
-    private static boolean probeAclSupport(DefaultMQAdminExt admin) {
+    private static boolean probeAclSupport(DefaultMQAdminExt admin, ClusterInfo clusterInfo) {
         try {
-            ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
+            if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+                return false;
+            }
             for (BrokerData broker : clusterInfo.getBrokerAddrTable().values()) {
                 String brokerAddr = broker.selectBrokerAddr();
                 if (brokerAddr == null || brokerAddr.isBlank()) {
@@ -2101,22 +2269,30 @@ public final class RocketMqAgent {
         if (!configured.isBlank()) {
             return configured;
         }
-        if (clusterInfo.getClusterAddrTable() != null && !clusterInfo.getClusterAddrTable().isEmpty()) {
+        if (clusterInfo != null
+            && clusterInfo.getClusterAddrTable() != null
+            && !clusterInfo.getClusterAddrTable().isEmpty()) {
             return clusterInfo.getClusterAddrTable().keySet().iterator().next();
         }
         return "DefaultCluster";
     }
 
     private static String resolveBrokerAddr(DefaultMQAdminExt admin, JsonObject conn) throws Exception {
+        return resolveBrokerAddr(admin, conn, admin.examineBrokerClusterInfo());
+    }
+
+    private static String resolveBrokerAddr(DefaultMQAdminExt admin, JsonObject conn, ClusterInfo clusterInfo)
+        throws Exception {
         String explicit = brokerAddress(conn);
         if (!explicit.isBlank()) {
             return explicit;
         }
-        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
-        for (BrokerData broker : clusterInfo.getBrokerAddrTable().values()) {
-            String addr = broker.selectBrokerAddr();
-            if (addr != null && !addr.isBlank()) {
-                return remapBrokerAddrForClient(addr, conn);
+        if (clusterInfo != null && clusterInfo.getBrokerAddrTable() != null) {
+            for (BrokerData broker : clusterInfo.getBrokerAddrTable().values()) {
+                String addr = broker.selectBrokerAddr();
+                if (addr != null && !addr.isBlank()) {
+                    return remapBrokerAddrForClient(addr, conn);
+                }
             }
         }
         throw new IllegalStateException("No RocketMQ broker address found");
@@ -2364,18 +2540,23 @@ public final class RocketMqAgent {
     }
 
     private static Set<String> collectBrokerNames(DefaultMQAdminExt admin) {
-        Set<String> names = new HashSet<>();
         try {
-            ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
-            if (clusterInfo.getBrokerAddrTable() != null) {
-                for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
-                    if (brokerData.getBrokerName() != null && !brokerData.getBrokerName().isBlank()) {
-                        names.add(brokerData.getBrokerName());
-                    }
-                }
-            }
+            return collectBrokerNames(admin.examineBrokerClusterInfo());
         } catch (Exception ignored) {
             // Fall back to static reserved-topic filtering only.
+            return Set.of();
+        }
+    }
+
+    static Set<String> collectBrokerNames(ClusterInfo clusterInfo) {
+        Set<String> names = new HashSet<>();
+        if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+            return names;
+        }
+        for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
+            if (brokerData.getBrokerName() != null && !brokerData.getBrokerName().isBlank()) {
+                names.add(brokerData.getBrokerName());
+            }
         }
         return names;
     }
@@ -2397,29 +2578,44 @@ public final class RocketMqAgent {
     }
 
     private static Map<String, TopicConfig> collectBrokerTopicConfigs(DefaultMQAdminExt admin, JsonObject conn) {
-        Map<String, TopicConfig> merged = new LinkedHashMap<>();
         try {
-            for (String brokerAddr : resolveMasterBrokerAddrs(admin, conn)) {
-                try {
-                    TopicConfigSerializeWrapper wrapper =
-                        admin.getAllTopicConfig(brokerAddr, DEFAULT_REQUEST_TIMEOUT_MS);
-                    if (wrapper == null || wrapper.getTopicConfigTable() == null) {
+            return collectBrokerTopicConfigSnapshot(admin, conn, admin.examineBrokerClusterInfo()).topics();
+        } catch (Exception ignored) {
+            return new LinkedHashMap<>();
+        }
+    }
+
+    private static BrokerTopicConfigSnapshot collectBrokerTopicConfigSnapshot(
+        DefaultMQAdminExt admin, JsonObject conn, ClusterInfo clusterInfo) throws Exception {
+        List<String> brokerAddrs = resolveMasterBrokerAddrs(admin, conn, null, clusterInfo);
+        return collectBrokerTopicConfigs(
+            brokerAddrs,
+            brokerAddr -> admin.getAllTopicConfig(brokerAddr, DEFAULT_REQUEST_TIMEOUT_MS)
+        );
+    }
+
+    static BrokerTopicConfigSnapshot collectBrokerTopicConfigs(
+        List<String> brokerAddrs, BrokerTopicConfigLookup lookup) {
+        Map<String, TopicConfig> merged = new LinkedHashMap<>();
+        boolean complete = !brokerAddrs.isEmpty();
+        for (String brokerAddr : brokerAddrs) {
+            try {
+                TopicConfigSerializeWrapper wrapper = lookup.load(brokerAddr);
+                if (wrapper == null || wrapper.getTopicConfigTable() == null) {
+                    complete = false;
+                    continue;
+                }
+                for (TopicConfig config : wrapper.getTopicConfigTable().values()) {
+                    if (config.getTopicName() == null || config.getTopicName().isBlank()) {
                         continue;
                     }
-                    for (TopicConfig config : wrapper.getTopicConfigTable().values()) {
-                        if (config.getTopicName() == null || config.getTopicName().isBlank()) {
-                            continue;
-                        }
-                        merged.merge(config.getTopicName(), config, RocketMqAgent::preferTopicConfig);
-                    }
-                } catch (Exception ignored) {
-                    // Some brokers may reject bulk config reads.
+                    merged.merge(config.getTopicName(), config, RocketMqAgent::preferTopicConfig);
                 }
+            } catch (Exception ignored) {
+                complete = false;
             }
-        } catch (Exception ignored) {
-            // Fall back to nameserver topic list in listTopics.
         }
-        return merged;
+        return new BrokerTopicConfigSnapshot(merged, complete);
     }
 
     private static TopicConfig preferTopicConfig(TopicConfig left, TopicConfig right) {
@@ -2477,13 +2673,24 @@ public final class RocketMqAgent {
     }
 
     private static String resolveTopicMessageType(DefaultMQAdminExt admin, JsonObject conn, String topic) {
+        try {
+            return resolveTopicMessageType(admin, conn, topic, admin.examineBrokerClusterInfo());
+        } catch (Exception ignored) {
+            return null;
+        }
+    }
+
+    private static String resolveTopicMessageType(
+        DefaultMQAdminExt admin, JsonObject conn, String topic, ClusterInfo clusterInfo) {
         try {
             TopicRouteData route = admin.examineTopicRouteInfo(topic);
             if (route.getQueueDatas() == null || route.getQueueDatas().isEmpty()) {
                 return null;
             }
             String brokerName = route.getQueueDatas().get(0).getBrokerName();
-            ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
+            if (clusterInfo == null || clusterInfo.getBrokerAddrTable() == null) {
+                return null;
+            }
             BrokerData brokerData = clusterInfo.getBrokerAddrTable().get(brokerName);
             if (brokerData == null) {
                 return null;
@@ -2499,10 +2706,17 @@ public final class RocketMqAgent {
         }
     }
 
+    /**
+     * Slice {@code items} from {@code offset}. When {@code limit <= 0}, return all remaining items
+     * (used by the Rust adapter's single-shot full list request).
+     */
     static <T> List<T> paginate(List<T> items, int offset, int limit) {
         if (offset >= items.size()) {
             return Collections.emptyList();
         }
+        if (limit <= 0) {
+            return new ArrayList<>(items.subList(offset, items.size()));
+        }
         int end = Math.min(items.size(), offset + limit);
         return new ArrayList<>(items.subList(offset, end));
     }
@@ -2514,9 +2728,12 @@ public final class RocketMqAgent {
         return adminClient;
     }
 
-    private static DefaultMQProducer requireProducer() {
+    private static DefaultMQProducer requireProducer() throws Exception {
         if (producer == null) {
-            throw new IllegalStateException("Producer is not initialized. Call connect first.");
+            if (cachedConnection == null) {
+                throw new IllegalStateException("Not connected. Call connect first.");
+            }
+            producer = buildProducer(cachedConnection);
         }
         return producer;
     }

agents/drivers/rocketmq/src/test/java/com/dbx/agent/rocketmq/RocketMqAgentTest.java

@@ -14,17 +14,23 @@ import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
 import org.apache.rocketmq.remoting.protocol.body.ProducerInfo;
 import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
 import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.common.message.MessageQueue;
 import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -180,6 +186,29 @@ class RocketMqAgentTest {
         assertEquals(200, RocketMqAgent.paginate(items, 200, 200).get(0));
     }
 
+    @Test
+    void paginateLimitZeroReturnsAllFromOffset() {
+        List<Integer> items = IntStream.range(0, 341).boxed().toList();
+        assertEquals(341, RocketMqAgent.paginate(items, 0, 0).size());
+        assertEquals(141, RocketMqAgent.paginate(items, 200, 0).size());
+        assertEquals(200, RocketMqAgent.paginate(items, 200, 0).get(0));
+    }
+
+    @Test
+    void collectBrokerNamesFromClusterInfoSnapshot() {
+        org.apache.rocketmq.remoting.protocol.body.ClusterInfo clusterInfo =
+            new org.apache.rocketmq.remoting.protocol.body.ClusterInfo();
+        java.util.HashMap<String, org.apache.rocketmq.remoting.protocol.route.BrokerData> table =
+            new java.util.HashMap<>();
+        org.apache.rocketmq.remoting.protocol.route.BrokerData broker =
+            new org.apache.rocketmq.remoting.protocol.route.BrokerData();
+        broker.setBrokerName("broker-a");
+        table.put("broker-a", broker);
+        clusterInfo.setBrokerAddrTable(table);
+        assertEquals(Set.of("broker-a"), RocketMqAgent.collectBrokerNames(clusterInfo));
+        assertEquals(Set.of(), RocketMqAgent.collectBrokerNames(null));
+    }
+
     @Test
     void resolveNameServerAddrSetSplitsMultiAddr() {
         JsonObject conn = JsonParser.parseString("""
@@ -292,6 +321,187 @@ class RocketMqAgentTest {
         }
     }
 
+    @Test
+    void enrichTopicPartitionsRunsIndependentLookupsConcurrently() throws Exception {
+        List<Map<String, Object>> rows = IntStream.range(0, 4)
+            .mapToObj(index -> {
+                Map<String, Object> row = new LinkedHashMap<>();
+                row.put("name", "topic-" + index);
+                row.put("partitions", 1);
+                return row;
+            })
+            .toList();
+        CountDownLatch started = new CountDownLatch(rows.size());
+        CountDownLatch release = new CountDownLatch(1);
+        AtomicInteger active = new AtomicInteger();
+        AtomicInteger maxActive = new AtomicInteger();
+
+        CompletableFuture<Void> enrichment = CompletableFuture.runAsync(() ->
+            RocketMqAgent.enrichTopicPartitions(rows, topic -> {
+                int current = active.incrementAndGet();
+                maxActive.accumulateAndGet(current, Math::max);
+                started.countDown();
+                try {
+                    release.await(1, TimeUnit.SECONDS);
+                    return routeWithReadQueues(8);
+                } finally {
+                    active.decrementAndGet();
+                }
+            }, 4, 1_000)
+        );
+
+        assertTrue(started.await(1, TimeUnit.SECONDS));
+        release.countDown();
+        enrichment.get(2, TimeUnit.SECONDS);
+
+        assertTrue(maxActive.get() > 1);
+        for (Map<String, Object> row : rows) {
+            assertEquals(8, row.get("partitions"));
+        }
+    }
+
+    @Test
+    void enrichTopicPartitionsReturnsDefaultsWhenBudgetExpires() {
+        List<Map<String, Object>> rows = IntStream.range(0, 4)
+            .mapToObj(index -> {
+                Map<String, Object> row = new LinkedHashMap<>();
+                row.put("name", "slow-topic-" + index);
+                row.put("partitions", 1);
+                return row;
+            })
+            .toList();
+        CountDownLatch blocked = new CountDownLatch(1);
+        long startedAt = System.nanoTime();
+
+        RocketMqAgent.enrichTopicPartitions(rows, topic -> {
+            blocked.await();
+            return routeWithReadQueues(16);
+        }, 2, 50);
+
+        long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedAt);
+        assertTrue(elapsedMs < 1_000, "topic route enrichment exceeded its response budget: " + elapsedMs + "ms");
+        for (Map<String, Object> row : rows) {
+            assertEquals(1, row.get("partitions"));
+        }
+    }
+
+    @Test
+    void buildTopicCatalogRowsFallbackIssuesAtMostOneRouteRpcPerTopic() {
+        Set<String> topicNames = new TreeSet<>(List.of("Alpha", "Beta", "Gamma"));
+        Map<String, AtomicInteger> routeCalls = new ConcurrentHashMap<>();
+        AtomicInteger totalRouteCalls = new AtomicInteger();
+
+        List<Map<String, Object>> rows = RocketMqAgent.buildTopicCatalogRows(
+            topicNames,
+            Collections.emptyMap(),
+            Collections.emptyMap(),
+            Set.of(),
+            Set.of(),
+            "DefaultCluster",
+            "",
+            topic -> {
+                totalRouteCalls.incrementAndGet();
+                routeCalls.computeIfAbsent(topic, ignored -> new AtomicInteger()).incrementAndGet();
+                return routeWithReadQueues(4);
+            },
+            8,
+            2_000
+        );
+
+        assertEquals(3, rows.size());
+        assertEquals(3, totalRouteCalls.get(), "fallback must not repeat examineTopicRouteInfo per topic");
+        for (String topic : topicNames) {
+            assertEquals(1, routeCalls.get(topic).get());
+        }
+        for (Map<String, Object> row : rows) {
+            assertEquals(4, row.get("partitions"));
+            // Bulk config unavailable — classify without per-topic examineTopicConfig.
+            assertEquals("UNSPECIFIED", row.get("messageType"));
+        }
+    }
+
+    @Test
+    void buildTopicCatalogRowsUsesBrokerConfigWithoutRouteLookup() {
+        TopicConfig config = new TopicConfig("Orders");
+        config.setReadQueueNums(12);
+        config.setAttributes(Map.of("message.type", "NORMAL"));
+        Map<String, TopicConfig> brokerTopics = Map.of("Orders", config);
+        AtomicInteger routeCalls = new AtomicInteger();
+
+        List<Map<String, Object>> rows = RocketMqAgent.buildTopicCatalogRows(
+            Set.of("Orders"),
+            brokerTopics,
+            Map.of("Orders", Map.of("message.type", "NORMAL")),
+            Set.of(),
+            Set.of(),
+            "DefaultCluster",
+            "",
+            topic -> {
+                routeCalls.incrementAndGet();
+                return routeWithReadQueues(99);
+            },
+            8,
+            2_000
+        );
+
+        assertEquals(1, rows.size());
+        assertEquals(0, routeCalls.get(), "broker catalog path must not call examineTopicRouteInfo");
+        assertEquals(12, rows.get(0).get("partitions"));
+        assertEquals("NORMAL", rows.get(0).get("messageType"));
+    }
+
+    @Test
+    void partialBrokerTopicConfigsMergeNameServerCatalog() {
+        TopicConfig orders = new TopicConfig("Orders");
+        TopicConfigSerializeWrapper wrapper = new TopicConfigSerializeWrapper();
+        wrapper.setTopicConfigTable(new ConcurrentHashMap<>(Map.of("Orders", orders)));
+
+        RocketMqAgent.BrokerTopicConfigSnapshot snapshot = RocketMqAgent.collectBrokerTopicConfigs(
+            List.of("broker-a", "broker-b"),
+            brokerAddr -> {
+                if (brokerAddr.equals("broker-b")) {
+                    throw new IllegalStateException("broker unavailable");
+                }
+                return wrapper;
+            }
+        );
+
+        assertFalse(snapshot.complete());
+        assertEquals(Set.of("Orders"), snapshot.topics().keySet());
+        assertEquals(
+            Set.of("Invoices", "Orders"),
+            RocketMqAgent.topicCatalogNames(snapshot, Set.of("Invoices", "Orders"))
+        );
+    }
+
+    @Test
+    void completeBrokerTopicConfigsRemainAuthoritative() {
+        TopicConfig orders = new TopicConfig("Orders");
+        TopicConfigSerializeWrapper wrapper = new TopicConfigSerializeWrapper();
+        wrapper.setTopicConfigTable(new ConcurrentHashMap<>(Map.of("Orders", orders)));
+
+        RocketMqAgent.BrokerTopicConfigSnapshot snapshot = RocketMqAgent.collectBrokerTopicConfigs(
+            List.of("broker-a"),
+            brokerAddr -> wrapper
+        );
+
+        assertTrue(snapshot.complete());
+        assertEquals(
+            Set.of("Orders"),
+            RocketMqAgent.topicCatalogNames(snapshot, Set.of("DeletedButStillRouted", "Orders"))
+        );
+    }
+
+    private static TopicRouteData routeWithReadQueues(int readQueueNums) {
+        QueueData queueData = new QueueData();
+        queueData.setBrokerName("broker-a");
+        queueData.setReadQueueNums(readQueueNums);
+        queueData.setWriteQueueNums(readQueueNums);
+        TopicRouteData route = new TopicRouteData();
+        route.setQueueDatas(List.of(queueData));
+        return route;
+    }
+
     @Test
     void isEmptyQueryMessageResultDetectsRocketMqCode208() {
         MQClientException empty = new MQClientException(208, "query message by key finished, but no message");
@@ -346,4 +556,23 @@ class RocketMqAgentTest {
         assertEquals("CLIENT_INNER_PRODUCER", producers.get(0).get("producerName"));
         assertEquals("127.0.0.1:39688", producers.get(0).get("address"));
     }
+
+    @Test
+    void connectionMatchesComparesRocketMqConnectFields() {
+        JsonObject base = JsonParser.parseString("""
+            {
+              "namesrv_addr": "127.0.0.1:9876",
+              "cluster_name": "DefaultCluster",
+              "broker_addr": "",
+              "access_key": "ak",
+              "secret_key": "sk"
+            }
+            """).getAsJsonObject();
+        JsonObject same = base.deepCopy();
+        JsonObject differentNamesrv = base.deepCopy();
+        differentNamesrv.addProperty("namesrv_addr", "127.0.0.1:9877");
+
+        assertTrue(RocketMqAgent.connectionMatches(base, same));
+        assertFalse(RocketMqAgent.connectionMatches(base, differentNamesrv));
+    }
 }

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

@@ -158,20 +158,25 @@ async function runQuery() {
       if (!id) throw new Error(t("mqMessages.msgIdRequired"));
       const result = await mqViewMessage(props.connectionId, topic, id);
       queryMessages.value = parseRocketMqMessagesFromResult(result);
+    } else if (activeQueryMode.value === "key") {
+      const key = queryKey.value.trim();
+      if (!key) throw new Error(t("mqMessages.queryKeyRequired"));
+      // Dashboard key query only needs topic + key; broker returns up to 64 recent matches.
+      // Do not validate hidden Topic-mode time fields — Key uses a fixed 0..now window.
+      const result = await mqQueryMessagesByKey(props.connectionId, topic, key, 0, Date.now(), 64);
+      queryMessages.value = parseRocketMqMessagesFromResult(result);
     } else {
-      const begin = queryBeginTime.value ? Date.parse(queryBeginTime.value) : 0;
-      const end = queryEndTime.value ? Date.parse(queryEndTime.value) : Date.now();
-      const maxNum = Math.max(1, Math.min(200, Number(queryMaxNum.value) || 32));
-      if (activeQueryMode.value === "key") {
-        const key = queryKey.value.trim();
-        if (!key) throw new Error(t("mqMessages.queryKeyRequired"));
-        // Dashboard key query only needs topic + key; broker returns up to 64 recent matches.
-        const result = await mqQueryMessagesByKey(props.connectionId, topic, key, 0, Date.now(), 64);
-        queryMessages.value = parseRocketMqMessagesFromResult(result);
-      } else {
-        const result = await mqQueryMessagesByTopic(props.connectionId, topic, begin, end, maxNum);
-        queryMessages.value = parseRocketMqMessagesFromResult(result);
+      const begin = queryBeginTime.value ? Date.parse(queryBeginTime.value) : Date.parse(defaultTimeRange().begin);
+      const end = queryEndTime.value ? Date.parse(queryEndTime.value) : Date.parse(defaultTimeRange().end);
+      if (!Number.isFinite(begin) || !Number.isFinite(end)) {
+        throw new Error(t("mqMessages.invalidTimeRange"));
       }
+      if (begin >= end) {
+        throw new Error(t("mqMessages.endTimeMustBeAfterBegin"));
+      }
+      const maxNum = Math.max(1, Math.min(200, Number(queryMaxNum.value) || 32));
+      const result = await mqQueryMessagesByTopic(props.connectionId, topic, begin, end, maxNum);
+      queryMessages.value = parseRocketMqMessagesFromResult(result);
     }
   } catch (e: unknown) {
     queryError.value = formatError(e);
@@ -401,20 +406,22 @@ watch(topicName, () => {
           <input v-model="queryKey" type="text" :placeholder="t('mqMessages.queryKeyPlaceholder')" :disabled="queryLoading" />
         </label>
 
-        <label v-else-if="activeQueryMode === 'topic'" class="filter-field">
-          <span>{{ t("mqMessages.beginTime") }}</span>
-          <input v-model="queryBeginTime" type="datetime-local" :disabled="queryLoading" />
-        </label>
+        <template v-else-if="activeQueryMode === 'topic'">
+          <label class="filter-field">
+            <span>{{ t("mqMessages.beginTime") }}</span>
+            <input v-model="queryBeginTime" type="datetime-local" :disabled="queryLoading" />
+          </label>
 
-        <label v-else-if="activeQueryMode === 'topic'" class="filter-field">
-          <span>{{ t("mqMessages.endTime") }}</span>
-          <input v-model="queryEndTime" type="datetime-local" :disabled="queryLoading" />
-        </label>
+          <label class="filter-field">
+            <span>{{ t("mqMessages.endTime") }}</span>
+            <input v-model="queryEndTime" type="datetime-local" :disabled="queryLoading" />
+          </label>
 
-        <label v-else-if="activeQueryMode === 'topic'" class="filter-field filter-narrow">
-          <span>{{ t("mqMessages.maxNum") }}</span>
-          <input v-model.number="queryMaxNum" type="number" min="1" max="200" :disabled="queryLoading" />
-        </label>
+          <label class="filter-field filter-narrow">
+            <span>{{ t("mqMessages.maxNum") }}</span>
+            <input v-model.number="queryMaxNum" type="number" min="1" max="200" :disabled="queryLoading" />
+          </label>
+        </template>
 
         <div class="filter-actions">
           <button type="button" class="btn-primary" :disabled="queryLoading || !selectedTopicRef" @click="runQuery">

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

@@ -1,6 +1,8 @@
 <script setup lang="ts">
 import { ref, watch, computed } from "vue";
 import { useI18n } from "vue-i18n";
+import { RecycleScroller } from "vue-virtual-scroller";
+import "vue-virtual-scroller/dist/vue-virtual-scroller.css";
 import type { NamespaceRef, TopicRef, TopicInfo, ListTopicsOpts, MqSystemKind, RocketMqTopicMessageType } from "@/types/mq";
 import { mqListTopics, mqCreateTopic, mqDeleteTopic, mqUpdatePartitions, mqGetClusterInfo } from "@/lib/backend/api";
 import type { ClusterInfo } from "@/types/mq";
@@ -14,6 +16,13 @@ import { formatError } from "@/lib/backend/errorUtils";
 import { DEFAULT_ROCKETMQ_TOPIC_TYPE_FILTERS, isProtectedRocketMqTopic, isRocketMqBusinessMessageType, matchesRocketMqTypeFilters, resolveRocketMqMessageType, ROCKETMQ_CREATABLE_TOPIC_MESSAGE_TYPES, ROCKETMQ_TOPIC_MESSAGE_TYPES } from "@/lib/mq/rocketmqTopicTypes";
 import DangerConfirmDialog from "@/components/editor/DangerConfirmDialog.vue";
 
+const TOPIC_ROW_HEIGHT = 44;
+
+type VirtualTopicRow = {
+  id: string;
+  topic: TopicInfo;
+};
+
 interface Props {
   connectionId: string;
   tenant?: string;
@@ -110,6 +119,38 @@ const filteredTopics = computed(() => {
   });
 });
 
+/** Stable ids for RecycleScroller; avoids mounting all topic rows at once. */
+const virtualTopicRows = computed<VirtualTopicRow[]>(() =>
+  filteredTopics.value.map((topic) => ({
+    id: showNamespaceColumn.value ? `${topic.namespace ?? ""}:${topic.name}` : topic.name,
+    topic,
+  })),
+);
+
+const topicsGridTemplate = computed(() => {
+  const cols: string[] = ["minmax(180px, 1.6fr)"];
+  if (showNamespaceColumn.value) cols.push("minmax(100px, 0.8fr)");
+  cols.push("120px");
+  if (!isRocketMqCluster.value) cols.push("140px");
+  // RocketMQ keeps tiled row actions (status/route/consumers/…) — reserve a wider actions column.
+  cols.push(isRocketMqCluster.value ? "minmax(560px, 2.2fr)" : "minmax(200px, 1fr)");
+  return cols.join(" ");
+});
+
+/** Min content width from grid track mins + gaps + padding; enables shared horizontal scroll. */
+const topicsTableMinWidthPx = computed(() => {
+  let min = 180 + 120; // name + type
+  if (showNamespaceColumn.value) min += 100;
+  if (!isRocketMqCluster.value) min += 140;
+  min += isRocketMqCluster.value ? 560 : 200;
+  let colCount = 3;
+  if (showNamespaceColumn.value) colCount += 1;
+  if (!isRocketMqCluster.value) colCount += 1;
+  min += (colCount - 1) * 8; // column-gap
+  min += 24; // horizontal padding
+  return min;
+});
+
 const userTopicCount = computed(() => {
   if (isRocketMqCluster.value) {
     return topics.value.filter((topic) => isRocketMqBusinessMessageType(resolveRocketMqMessageType(topic))).length;
@@ -117,6 +158,10 @@ const userTopicCount = computed(() => {
   return topics.value.filter((topic) => !topic.internal).length;
 });
 
+function topicRowSelected(topic: TopicInfo): boolean {
+  return selectedTopic.value?.name === topic.name && selectedTopic.value?.namespace === topic.namespace;
+}
+
 function topicTypeLabel(topic: TopicInfo): string {
   if (isRocketMqCluster.value) {
     const type = resolveRocketMqMessageType(topic);
@@ -470,61 +515,68 @@ watch(newPartitions, () => {
       </div>
 
       <div v-else class="topics-table">
-        <table>
-          <thead>
-            <tr>
-              <th>{{ t("mqTopics.name") }}</th>
-              <th v-if="showNamespaceColumn">{{ t("mqAdmin.namespace") }}</th>
-              <th>{{ t("mqTopics.type") }}</th>
-              <th v-if="!isRocketMqCluster">{{ t("mqTopics.partitions") }}</th>
-              <th>{{ t("mqTopics.actions") }}</th>
-            </tr>
-          </thead>
-          <tbody>
-            <tr v-for="topic in filteredTopics" :key="showNamespaceColumn ? `${topic.namespace ?? ''}:${topic.name}` : topic.name" :class="{ selected: selectedTopic?.name === topic.name && selectedTopic?.namespace === topic.namespace }" @click="selectTopic(topic)">
-              <td class="topic-name">
-                <div class="topic-name-cell">
-                  <span>{{ topic.shortName }}</span>
-                  <span v-if="!topic.persistent" class="badge badge-warning">{{ t("mqTopics.nonPersistent") }}</span>
+        <!-- Shared horizontal scroller keeps header/body columns aligned when the grid min-width exceeds the panel. -->
+        <div class="topics-table-hscroll" :style="{ '--topics-table-min-width': `${topicsTableMinWidthPx}px` }">
+          <div class="topics-table-header" :style="{ gridTemplateColumns: topicsGridTemplate }">
+            <div class="topics-col">{{ t("mqTopics.name") }}</div>
+            <div v-if="showNamespaceColumn" class="topics-col">{{ t("mqAdmin.namespace") }}</div>
+            <div class="topics-col">{{ t("mqTopics.type") }}</div>
+            <div v-if="!isRocketMqCluster" class="topics-col">{{ t("mqTopics.partitions") }}</div>
+            <div class="topics-col">{{ t("mqTopics.actions") }}</div>
+          </div>
+          <RecycleScroller class="topics-scroller" :items="virtualTopicRows" :item-size="TOPIC_ROW_HEIGHT" :buffer="200" key-field="id">
+            <template #default="{ item: row }">
+              <div class="topics-row" :class="{ selected: topicRowSelected(row.topic) }" :style="{ gridTemplateColumns: topicsGridTemplate, height: `${TOPIC_ROW_HEIGHT}px` }" @click="selectTopic(row.topic)">
+                <div class="topics-col topic-name">
+                  <div class="topic-name-cell">
+                    <span class="topic-name-text" :title="row.topic.shortName">{{ row.topic.shortName }}</span>
+                    <span v-if="!row.topic.persistent" class="badge badge-warning">{{ t("mqTopics.nonPersistent") }}</span>
+                  </div>
                 </div>
-              </td>
-              <td v-if="showNamespaceColumn">{{ topic.namespace || "-" }}</td>
-              <td>
-                <span class="badge" :class="topicTypeBadgeClass(topic)">
-                  {{ topicTypeLabel(topic) }}
-                </span>
-              </td>
-              <td v-if="!isRocketMqCluster">
-                <span v-if="topic.partitioned">{{ topic.partitions ? t("mqTopics.partitionCount", { count: topic.partitions }) : t("mqTopics.partitionsUnknown") }}</span>
-                <span v-else class="text-muted">-</span>
-              </td>
-              <td class="actions" @click.stop>
-                <template v-if="isRocketMqCluster">
-                  <button class="btn-sm" @click="openRocketMqDialog('status', topic)">{{ t("mqTopics.actionStatus") }}</button>
-                  <button class="btn-sm" @click="openRocketMqDialog('route', topic)">{{ t("mqTopics.actionRoute") }}</button>
-                  <button class="btn-sm" @click="openRocketMqDialog('consumers', topic)">{{ t("mqTopics.actionConsumers") }}</button>
-                  <button v-if="isDlqTopic(topic)" class="btn-sm" @click="navigateToMessageQuery(topic, true)">{{ t("mqRocketmq.viewDlqMessages") }}</button>
-                  <template v-else>
-                    <button class="btn-sm" @click="navigateToMessageQuery(topic)">{{ t("mqRocketmq.actionMessageQuery") }}</button>
-                    <button class="btn-sm" :disabled="readOnly" @click="navigateToMessages(topic)">{{ t("mqRocketmq.actionSendMessage") }}</button>
+                <div v-if="showNamespaceColumn" class="topics-col">{{ row.topic.namespace || "-" }}</div>
+                <div class="topics-col">
+                  <span class="badge" :class="topicTypeBadgeClass(row.topic)">
+                    {{ topicTypeLabel(row.topic) }}
+                  </span>
+                </div>
+                <div v-if="!isRocketMqCluster" class="topics-col">
+                  <span v-if="row.topic.partitioned">{{ row.topic.partitions ? t("mqTopics.partitionCount", { count: row.topic.partitions }) : t("mqTopics.partitionsUnknown") }}</span>
+                  <span v-else class="text-muted">-</span>
+                </div>
+                <div class="topics-col actions" @click.stop>
+                  <template v-if="isRocketMqCluster">
+                    <button type="button" class="btn-sm" @click="openRocketMqDialog('status', row.topic)">{{ t("mqTopics.actionStatus") }}</button>
+                    <button type="button" class="btn-sm" @click="openRocketMqDialog('route', row.topic)">{{ t("mqTopics.actionRoute") }}</button>
+                    <button type="button" class="btn-sm" @click="openRocketMqDialog('consumers', row.topic)">{{ t("mqTopics.actionConsumers") }}</button>
+                    <button v-if="isDlqTopic(row.topic)" type="button" class="btn-sm" @click="navigateToMessageQuery(row.topic, true)">{{ t("mqRocketmq.viewDlqMessages") }}</button>
+                    <template v-else>
+                      <button type="button" class="btn-sm" @click="navigateToMessageQuery(row.topic)">{{ t("mqRocketmq.actionMessageQuery") }}</button>
+                      <button type="button" class="btn-sm" :disabled="readOnly" @click="navigateToMessages(row.topic)">{{ t("mqRocketmq.actionSendMessage") }}</button>
+                    </template>
+                    <button type="button" class="btn-sm" @click="openRocketMqDialog('config', row.topic)">{{ t("mqTopics.actionConfig") }}</button>
+                    <button type="button" class="btn-sm" :disabled="readOnly || isTopicProtected(row.topic)" @click="openRocketMqDialog('reset', row.topic)">{{ t("mqTopics.actionReset") }}</button>
+                    <button type="button" class="btn-sm" :disabled="readOnly || isTopicProtected(row.topic)" @click="openRocketMqDialog('skip', row.topic)">{{ t("mqTopics.actionSkip") }}</button>
+                    <button type="button" class="btn-sm btn-danger" :disabled="readOnly || isTopicProtected(row.topic)" @click="handleDelete(row.topic)">{{ t("mqTopics.delete") }}</button>
                   </template>
-                  <button class="btn-sm" @click="openRocketMqDialog('config', topic)">{{ t("mqTopics.actionConfig") }}</button>
-                  <button class="btn-sm" :disabled="readOnly || isTopicProtected(topic)" @click="openRocketMqDialog('reset', topic)">{{ t("mqTopics.actionReset") }}</button>
-                  <button class="btn-sm" :disabled="readOnly || isTopicProtected(topic)" @click="openRocketMqDialog('skip', topic)">{{ t("mqTopics.actionSkip") }}</button>
-                  <button class="btn-sm btn-danger" :disabled="readOnly || isTopicProtected(topic)" @click="handleDelete(topic)">{{ t("mqTopics.delete") }}</button>
-                </template>
-                <template v-else>
-                  <button v-if="topic.partitioned && supportsPartitionedTopics !== false && !isTopicProtected(topic)" @click="openPartitionsDialog(topic)" :disabled="readOnly || !topic.partitions" class="btn-sm">
-                    {{ t("mqTopics.adjustPartitions") }}
-                  </button>
-                  <button @click="handleDelete(topic)" :disabled="readOnly || isTopicProtected(topic) || (showNamespaceColumn && !topic.namespace)" :title="showNamespaceColumn && !topic.namespace ? t('mqAdmin.selectNamespaceToWrite') : undefined" class="btn-sm btn-danger">
-                    {{ t("mqTopics.delete") }}
-                  </button>
-                </template>
-              </td>
-            </tr>
-          </tbody>
-        </table>
+                  <template v-else>
+                    <button v-if="row.topic.partitioned && supportsPartitionedTopics !== false && !isTopicProtected(row.topic)" type="button" @click="openPartitionsDialog(row.topic)" :disabled="readOnly || !row.topic.partitions" class="btn-sm">
+                      {{ t("mqTopics.adjustPartitions") }}
+                    </button>
+                    <button
+                      type="button"
+                      @click="handleDelete(row.topic)"
+                      :disabled="readOnly || isTopicProtected(row.topic) || (showNamespaceColumn && !row.topic.namespace)"
+                      :title="showNamespaceColumn && !row.topic.namespace ? t('mqAdmin.selectNamespaceToWrite') : undefined"
+                      class="btn-sm btn-danger"
+                    >
+                      {{ t("mqTopics.delete") }}
+                    </button>
+                  </template>
+                </div>
+              </div>
+            </template>
+          </RecycleScroller>
+        </div>
       </div>
 
       <!-- Create Dialog -->
@@ -809,87 +861,94 @@ watch(newPartitions, () => {
 .topics-table {
   position: relative;
   flex: 1;
-  overflow: auto;
+  min-height: 0;
+  display: flex;
+  flex-direction: column;
+  overflow: hidden;
   background: var(--topics-surface);
 }
 
-.topics-table::before {
-  content: "";
-  position: sticky;
-  top: 0;
-  display: block;
+.topics-table-hscroll {
+  flex: 1;
+  min-height: 0;
+  min-width: 0;
+  display: flex;
+  flex-direction: column;
+  overflow-x: auto;
+  overflow-y: hidden;
+}
+
+.topics-table-header,
+.topics-row {
+  display: grid;
+  align-items: center;
+  column-gap: 8px;
+  padding: 0 12px;
+  min-width: var(--topics-table-min-width, 100%);
+}
+
+.topics-table-header {
+  flex: 0 0 auto;
   height: 38px;
-  margin-bottom: -38px;
-  background: var(--topics-header-bg);
-  z-index: 9;
-  box-shadow:
-    0 1px 0 var(--topics-border),
-    0 2px 8px rgba(0, 0, 0, 0.05);
-  pointer-events: none;
-}
-
-table {
-  position: relative;
-  width: 100%;
-  border-collapse: separate;
-  border-spacing: 0;
-}
-
-thead {
-  position: sticky;
-  top: 0;
-  background: var(--topics-header-bg);
-  z-index: 10;
-}
-
-th {
-  position: sticky;
-  top: 0;
-  z-index: 11;
-  padding: 10px 12px;
-  text-align: left;
-  font-weight: 600;
-  font-size: 13px;
-  color: var(--color-text-secondary);
+  /* Match RecycleScroller classic-scrollbar gutter so header/body columns stay aligned. */
+  overflow: hidden;
+  scrollbar-gutter: stable;
   background: var(--topics-header-bg);
   border-bottom: 1px solid var(--topics-border);
-  background-clip: padding-box;
   box-shadow:
     0 1px 0 var(--topics-border),
     0 2px 6px rgba(0, 0, 0, 0.04);
+  z-index: 2;
 }
 
-tbody tr {
+.topics-table-header .topics-col {
+  font-weight: 600;
+  font-size: 13px;
+  color: var(--color-text-secondary);
+}
+
+.topics-scroller {
+  flex: 1;
+  min-height: 0;
+  height: 100%;
+  min-width: var(--topics-table-min-width, 100%);
+  /* RecycleScroller owns overflow-y; reserve gutter for classic scrollbars (Windows). */
+  scrollbar-gutter: stable;
+}
+
+.topics-row {
+  width: 100%;
   cursor: pointer;
-  transition: background 0.2s;
-}
-
-tbody tr:hover {
-  background: var(--color-hover);
-}
-
-tbody tr:hover td {
-  background: var(--color-hover);
-}
-
-tbody tr.selected {
-  background: var(--color-primary-alpha);
-}
-
-tbody tr.selected td {
-  background: var(--color-primary-alpha);
-}
-
-td {
-  padding: 10px 12px;
   border-bottom: 1px solid var(--topics-border-light);
   background: var(--topics-surface);
+  box-sizing: border-box;
+}
+
+.topics-row:hover {
+  background: var(--color-hover);
+}
+
+.topics-row.selected {
+  background: var(--color-primary-alpha);
+}
+
+.topics-col {
+  min-width: 0;
+  font-size: 13px;
 }
 
 .topic-name-cell {
   display: flex;
   align-items: center;
   gap: 8px;
+  min-width: 0;
+}
+
+.topic-name-text {
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+  font-weight: 500;
 }
 
 .topic-name {
@@ -927,9 +986,11 @@ td {
 .actions {
   display: flex;
   gap: 6px;
-  flex-wrap: wrap;
+  flex-wrap: nowrap;
   align-items: center;
-  min-width: 420px;
+  justify-content: flex-start;
+  overflow-x: auto;
+  min-width: 0;
 }
 
 .form-row-inline {

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

@@ -0,0 +1,142 @@
+// @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(),
+  mqQueryMessagesByKey: vi.fn(),
+  mqQueryMessagesByTopic: vi.fn(),
+  mqViewMessage: vi.fn(),
+  mqPeekMessages: vi.fn(),
+  mqSendMessage: vi.fn(),
+}));
+
+vi.mock("vue-i18n", () => ({
+  useI18n: () => ({ t: (key: string) => key }),
+}));
+
+vi.mock("@/lib/backend/api", () => ({
+  mqListTopics: backend.mqListTopics,
+  mqQueryMessagesByKey: backend.mqQueryMessagesByKey,
+  mqQueryMessagesByTopic: backend.mqQueryMessagesByTopic,
+  mqViewMessage: backend.mqViewMessage,
+  mqPeekMessages: backend.mqPeekMessages,
+  mqSendMessage: backend.mqSendMessage,
+}));
+
+import MessageQueryPanel from "@/components/mq/MessageQueryPanel.vue";
+
+const TOPIC = {
+  name: "Orders",
+  shortName: "Orders",
+  partitioned: true,
+  persistent: true,
+  messageType: "NORMAL",
+};
+
+let app: App<Element> | null = null;
+let root: HTMLDivElement | null = null;
+
+async function flushUi() {
+  await Promise.resolve();
+  await Promise.resolve();
+  await nextTick();
+  await Promise.resolve();
+  await nextTick();
+}
+
+function buttonByExactText(container: ParentNode, text: string): HTMLButtonElement {
+  const button = [...container.querySelectorAll<HTMLButtonElement>("button")].find((item) => item.textContent?.trim() === text);
+  if (!button) throw new Error(`Button not found: ${text}`);
+  return button;
+}
+
+async function setInputValue(input: HTMLInputElement | HTMLTextAreaElement, value: string) {
+  input.value = value;
+  input.dispatchEvent(new Event("input", { bubbles: true }));
+  await nextTick();
+}
+
+async function mountPanel() {
+  root = document.createElement("div");
+  document.body.appendChild(root);
+  app = createApp(MessageQueryPanel, {
+    connectionId: "mq-1",
+    tenant: "_rocketmq",
+    namespace: "default",
+    topic: TOPIC,
+    mqSystemKind: "rocketmq",
+    embedded: true,
+  });
+  app.config.globalProperties.$t = (key: string) => key;
+  app.mount(root);
+  await flushUi();
+  return root;
+}
+
+beforeEach(() => {
+  backend.mqListTopics.mockReset();
+  backend.mqQueryMessagesByKey.mockReset();
+  backend.mqQueryMessagesByTopic.mockReset();
+  backend.mqViewMessage.mockReset();
+  backend.mqPeekMessages.mockReset();
+  backend.mqSendMessage.mockReset();
+  backend.mqListTopics.mockResolvedValue([TOPIC]);
+  backend.mqQueryMessagesByKey.mockResolvedValue({ messages: [] });
+  backend.mqQueryMessagesByTopic.mockResolvedValue({ messages: [] });
+});
+
+afterEach(() => {
+  app?.unmount();
+  app = null;
+  root?.remove();
+  root = null;
+});
+
+async function waitForQueryEnabled(panel: HTMLElement) {
+  for (let attempt = 0; attempt < 20; attempt++) {
+    const queryButton = buttonByExactText(panel, "mqMessages.query");
+    if (!queryButton.disabled) return queryButton;
+    await flushUi();
+  }
+  throw new Error("Query button stayed disabled (topic not selected)");
+}
+
+describe("MessageQueryPanel Key mode time validation", () => {
+  it("does not block Key query with a stale invalid Topic time range", async () => {
+    const panel = await mountPanel();
+    await expect.poll(() => backend.mqListTopics.mock.calls.length).toBeGreaterThan(0);
+    await flushUi();
+
+    const beginInput = panel.querySelector<HTMLInputElement>('input[type="datetime-local"]');
+    const endInputs = panel.querySelectorAll<HTMLInputElement>('input[type="datetime-local"]');
+    if (!beginInput || endInputs.length < 2) throw new Error("Topic time inputs not found");
+    // Invalid range: begin after end (hidden after switching to Key).
+    await setInputValue(beginInput, "2026-08-03T18:00");
+    await setInputValue(endInputs[1], "2026-08-03T10:00");
+
+    buttonByExactText(panel, "mqMessages.queryTabKey").click();
+    await flushUi();
+
+    const keyInput = panel.querySelector<HTMLInputElement>('input[placeholder="mqMessages.queryKeyPlaceholder"]');
+    if (!keyInput) throw new Error("Key input not found");
+    await setInputValue(keyInput, "order-1");
+
+    const queryButton = await waitForQueryEnabled(panel);
+    queryButton.click();
+    await flushUi();
+    await expect.poll(() => backend.mqQueryMessagesByKey.mock.calls.length).toBe(1);
+
+    expect(panel.textContent).not.toContain("mqMessages.invalidTimeRange");
+    expect(panel.textContent).not.toContain("mqMessages.endTimeMustBeAfterBegin");
+    expect(backend.mqQueryMessagesByTopic).not.toHaveBeenCalled();
+    const [, topicRef, key, begin, end, maxNum] = backend.mqQueryMessagesByKey.mock.calls[0];
+    expect(topicRef).toMatchObject({ topic: "Orders" });
+    expect(key).toBe("order-1");
+    expect(begin).toBe(0);
+    expect(typeof end).toBe("number");
+    expect(end).toBeGreaterThan(0);
+    expect(maxNum).toBe(64);
+  });
+});

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

@@ -6052,6 +6052,8 @@ export default {
     beginTime: "Begin time",
     endTime: "End time",
     maxNum: "Max messages",
+    invalidTimeRange: "Begin time or end time is invalid",
+    endTimeMustBeAfterBegin: "End time must be later than begin time",
     query: "Query",
     querying: "Querying...",
     messageOverview: "Overview",

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

@@ -5910,6 +5910,8 @@ export default withEnglishFallback({
     beginTime: "Hora de inicio",
     endTime: "Hora de fin",
     maxNum: "Número máximo",
+    invalidTimeRange: "Hora de inicio o fin no válida",
+    endTimeMustBeAfterBegin: "La hora de fin debe ser posterior a la hora de inicio",
     query: "Consultar",
     querying: "Consultando...",
     messageOverview: "Información básica",

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

@@ -5910,6 +5910,8 @@ export default withEnglishFallback({
     beginTime: "Ora inizio",
     endTime: "Ora fine",
     maxNum: "Numero massimo",
+    invalidTimeRange: "Ora di inizio o fine non valida",
+    endTimeMustBeAfterBegin: "L'ora di fine deve essere successiva all'ora di inizio",
     query: "Cerca",
     querying: "Ricerca in corso...",
     messageOverview: "Informazioni di base",

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

@@ -5965,6 +5965,8 @@ export default withEnglishFallback({
     beginTime: "開始時間",
     endTime: "終了時間",
     maxNum: "最大件数",
+    invalidTimeRange: "開始時刻または終了時刻が無効です",
+    endTimeMustBeAfterBegin: "終了時刻は開始時刻より後である必要があります",
     query: "検索",
     querying: "検索中...",
     messageOverview: "基本情報",

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

@@ -5644,6 +5644,8 @@ export default withEnglishFallback({
     beginTime: "시작 시간",
     endTime: "종료 시간",
     maxNum: "최대 메시지 수",
+    invalidTimeRange: "시작 시간 또는 종료 시간이 유효하지 않습니다",
+    endTimeMustBeAfterBegin: "종료 시간은 시작 시간보다 늦어야 합니다",
     query: "쿼리",
     querying: "쿼리 중...",
     messageOverview: "개요",

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

@@ -5912,6 +5912,8 @@ export default withEnglishFallback({
     beginTime: "Hora inicial",
     endTime: "Hora final",
     maxNum: "Número máximo",
+    invalidTimeRange: "Hora inicial ou final inválida",
+    endTimeMustBeAfterBegin: "A hora final deve ser posterior à hora inicial",
     query: "Pesquisar",
     querying: "Pesquisando...",
     messageOverview: "Informações básicas",

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

@@ -6051,6 +6051,8 @@ export default withEnglishFallback({
     beginTime: "开始时间",
     endTime: "结束时间",
     maxNum: "最大条数",
+    invalidTimeRange: "开始时间或结束时间无效",
+    endTimeMustBeAfterBegin: "结束时间必须晚于开始时间",
     query: "查询",
     querying: "查询中...",
     messageOverview: "基本信息",

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

@@ -5905,6 +5905,8 @@ export default withEnglishFallback({
     beginTime: "開始時間",
     endTime: "結束時間",
     maxNum: "最大條數",
+    invalidTimeRange: "開始時間或結束時間無效",
+    endTimeMustBeAfterBegin: "結束時間必須晚於開始時間",
     query: "查詢",
     querying: "查詢中...",
     messageOverview: "基本資訊",

crates/dbx-core/src/connection.rs

@@ -2170,23 +2170,37 @@ impl AppState {
                 // connection_id is recognized as valid.
                 let mqc = self.mq_admin_config_for_connection(connection_id, &config).await?;
                 let agent_launch = crate::mq::service::resolve_mq_agent_launch_spec(&mqc, self);
-                let adapter = match self.mq_registry.get_or_build_config(connection_id, mqc, agent_launch).await {
-                    Ok(adapter) => adapter,
-                    Err(err) => {
+                // Temporary "__test_*" probes must not retain agents in the registry.
+                // reconnect fast-path caching only applies to durable connection ids;
+                // drain_connection_pools no longer drops MQ adapters (reconnect reuse).
+                if connection_id.starts_with("__test_") {
+                    let adapter = self.mq_registry.build_transient_config(mqc, agent_launch).await?;
+                    adapter.test_connection().await?;
+                    if let Err(err) = self.ensure_current_connection_attempt(connection_id, connection_attempt).await {
+                        self.reset_connection_transport_for_config(connection_id, &db_config).await;
+                        return Err(err);
+                    }
+                    // adapter drops here and kills agent-backed processes (RocketMQ/Kafka/RabbitMQ).
+                    PoolKind::MessageQueue
+                } else {
+                    let build = match self.mq_registry.get_or_build_config(connection_id, mqc, agent_launch).await {
+                        Ok(build) => build,
+                        Err(err) => {
+                            self.mq_registry.drop_connection(connection_id).await;
+                            return Err(err);
+                        }
+                    };
+                    if let Err(err) = crate::mq::validate_mq_adapter_after_build(&build).await {
                         self.mq_registry.drop_connection(connection_id).await;
                         return Err(err);
                     }
-                };
-                if let Err(err) = adapter.test_connection().await {
-                    self.mq_registry.drop_connection(connection_id).await;
-                    return Err(err);
+                    if let Err(err) = self.ensure_current_connection_attempt(connection_id, connection_attempt).await {
+                        self.mq_registry.drop_connection(connection_id).await;
+                        self.reset_connection_transport_for_config(connection_id, &db_config).await;
+                        return Err(err);
+                    }
+                    PoolKind::MessageQueue
                 }
-                if let Err(err) = self.ensure_current_connection_attempt(connection_id, connection_attempt).await {
-                    self.mq_registry.drop_connection(connection_id).await;
-                    self.reset_connection_transport_for_config(connection_id, &db_config).await;
-                    return Err(err);
-                }
-                PoolKind::MessageQueue
             }
             #[cfg(not(feature = "mq-admin"))]
             DatabaseType::MessageQueue => {
@@ -3943,9 +3957,6 @@ impl AppState {
             }
         }
         drop(conns);
-        // Also drop the MQ admin adapter if this is an MQ connection.
-        #[cfg(feature = "mq-admin")]
-        self.mq_registry.drop_connection(connection_id).await;
         removed
     }
 

crates/dbx-core/src/mq/adapters/rocketmq.rs

@@ -52,13 +52,46 @@ const ROCKETMQ_CAPABILITIES: MqCapabilities = MqCapabilities {
     supports_cluster_monitoring: false,
 };
 
-const TOPIC_LIST_PAGE_SIZE: u32 = 200;
+/// Prefer one RPC for the full topic catalog.
+///
+/// Use a large *positive* limit (not `0`): older agents coerce `limit <= 0` back to 200,
+/// which truncates catalogs (e.g. 338 topics → 200 rows → ~167 after type filters).
+/// Current agents treat `limit <= 0` as "all"; a large positive limit returns the same
+/// full page via normal pagination math without that version skew.
+const TOPIC_LIST_FETCH_LIMIT: i32 = i32::MAX;
+/// Fallback page size when an agent still returns a truncated first page (`topics.len() < total`).
+const TOPIC_LIST_FALLBACK_PAGE_SIZE: i32 = 200;
 
 pub struct RocketMqAdmin {
     client: Arc<Mutex<AgentDriverClient>>,
     config: MqAdminConfig,
 }
 
+fn cluster_info_from_agent_result(result: &serde_json::Value) -> MqClusterInfo {
+    let cluster_id = result.get("clusterId").and_then(|v| v.as_str()).map(String::from);
+    let brokers = result.get("brokers").cloned().unwrap_or(serde_json::json!([]));
+
+    // When the broker has no authorizer configured, disable permissions in the UI
+    // so the frontend hides the tab instead of showing raw errors.
+    let acl_enabled = result.get("aclEnabled").and_then(|v| v.as_bool()).unwrap_or(true);
+    let mut caps = ROCKETMQ_CAPABILITIES;
+    if !acl_enabled {
+        caps.supports_permissions = false;
+    }
+
+    MqClusterInfo {
+        system_kind: MqSystemKind::RocketMq,
+        server_version: None,
+        resolved_profile: "rocketmq-agent".to_string(),
+        version_detection: "agent".to_string(),
+        capabilities: caps,
+        extra: serde_json::json!({
+            "clusterId": cluster_id,
+            "brokers": brokers,
+        }),
+    }
+}
+
 impl RocketMqAdmin {
     /// Spawn the RocketMQ Java agent, perform handshake, and connect.
     pub async fn new(cfg: MqAdminConfig, launch: AgentLaunchSpec) -> Result<Self, String> {
@@ -104,33 +137,16 @@ impl MessageQueueAdmin for RocketMqAdmin {
         MqSystemKind::RocketMq
     }
 
+    fn build_includes_connect_test(&self) -> bool {
+        true
+    }
+
     async fn test_connection(&self) -> Result<MqClusterInfo, String> {
         let conn_params = build_connection_params(&self.config);
         let result: serde_json::Value =
             self.call("test_connection", serde_json::json!({ "connection": conn_params })).await?;
 
-        let cluster_id = result.get("clusterId").and_then(|v| v.as_str()).map(String::from);
-        let brokers = result.get("brokers").cloned().unwrap_or(serde_json::json!([]));
-
-        // When the broker has no authorizer configured, disable permissions in the UI
-        // so the frontend hides the tab instead of showing raw errors.
-        let acl_enabled = result.get("aclEnabled").and_then(|v| v.as_bool()).unwrap_or(true);
-        let mut caps = ROCKETMQ_CAPABILITIES;
-        if !acl_enabled {
-            caps.supports_permissions = false;
-        }
-
-        Ok(MqClusterInfo {
-            system_kind: MqSystemKind::RocketMq,
-            server_version: None,
-            resolved_profile: "rocketmq-agent".to_string(),
-            version_detection: "agent".to_string(),
-            capabilities: caps,
-            extra: serde_json::json!({
-                "clusterId": cluster_id,
-                "brokers": brokers,
-            }),
-        })
+        Ok(cluster_info_from_agent_result(&result))
     }
 
     // ---- Tenants (not supported by RocketMQ) ----
@@ -176,35 +192,40 @@ impl MessageQueueAdmin for RocketMqAdmin {
     // ---- Topics ----
 
     async fn list_topics(&self, _ns: &NamespaceRef, _opts: ListTopicsOpts) -> Result<Vec<TopicInfo>, String> {
-        let mut all = Vec::new();
-        let mut offset: u32 = 0;
+        // Prefer a single RPC so the agent builds the catalog once. If the agent still
+        // truncates (old coerce-to-200 behavior), page the remainder using `total`.
+        let result: serde_json::Value = self
+            .call(
+                "mq_list_topics",
+                serde_json::json!({
+                    "keyword": "",
+                    "limit": TOPIC_LIST_FETCH_LIMIT,
+                    "offset": 0,
+                }),
+            )
+            .await?;
 
-        loop {
-            let result: serde_json::Value = self
+        let mut all = topic_infos_from_agent_list_response(&result);
+        let total = agent_list_total(&result, all.len());
+        let mut offset = all.len();
+        while (offset as u64) < total {
+            let page: serde_json::Value = self
                 .call(
                     "mq_list_topics",
                     serde_json::json!({
                         "keyword": "",
-                        "limit": TOPIC_LIST_PAGE_SIZE,
+                        "limit": TOPIC_LIST_FALLBACK_PAGE_SIZE,
                         "offset": offset,
                     }),
                 )
                 .await?;
-
-            let topics = result.get("topics").and_then(|v| v.as_array()).cloned().unwrap_or_default();
-            let page_len = topics.len();
-            for topic in topics {
-                all.push(topic_info_from_agent_value(&topic));
-            }
-
-            let total = result.get("total").and_then(|v| v.as_u64()).unwrap_or(offset as u64 + page_len as u64);
-            let fetch_next = topic_list_should_fetch_next(offset, page_len, total);
-            offset = offset.saturating_add(page_len as u32);
-            if !fetch_next {
+            let batch = topic_infos_from_agent_list_response(&page);
+            if batch.is_empty() {
                 break;
             }
+            offset = offset.saturating_add(batch.len());
+            all.extend(batch);
         }
-
         Ok(all)
     }
 
@@ -735,28 +756,13 @@ fn topic_info_from_agent_value(t: &serde_json::Value) -> TopicInfo {
     }
 }
 
-/// Whether another Agent topic list page should be requested after consuming `page_len` rows.
-fn topic_list_should_fetch_next(offset: u32, page_len: usize, total: u64) -> bool {
-    page_len > 0 && u64::from(offset) + (page_len as u64) < total
+fn topic_infos_from_agent_list_response(response: &serde_json::Value) -> Vec<TopicInfo> {
+    response.get("topics").and_then(|v| v.as_array()).into_iter().flatten().map(topic_info_from_agent_value).collect()
 }
 
-#[cfg(test)]
-fn topic_infos_from_agent_pages(pages: &[serde_json::Value]) -> Vec<TopicInfo> {
-    let mut all = Vec::new();
-    let mut offset: u32 = 0;
-    for page in pages {
-        let topics = page.get("topics").and_then(|v| v.as_array()).cloned().unwrap_or_default();
-        let page_len = topics.len();
-        for topic in topics {
-            all.push(topic_info_from_agent_value(&topic));
-        }
-        let total = page.get("total").and_then(|v| v.as_u64()).unwrap_or(offset as u64 + page_len as u64);
-        offset = offset.saturating_add(page_len as u32);
-        if page_len == 0 || offset as u64 >= total {
-            break;
-        }
-    }
-    all
+/// Prefer agent-reported `total`; fall back to the page length when absent.
+fn agent_list_total(response: &serde_json::Value, page_len: usize) -> u64 {
+    response.get("total").and_then(|v| v.as_u64()).unwrap_or(page_len as u64)
 }
 
 /// Extract RocketMQ NameServer address from MqAdminConfig.extra.
@@ -1043,40 +1049,28 @@ mod tests {
     }
 
     #[test]
-    fn topic_list_should_fetch_next_when_more_rows_remain() {
-        assert!(!topic_list_should_fetch_next(200, 0, 201));
-        assert!(topic_list_should_fetch_next(0, 200, 201));
-        assert!(!topic_list_should_fetch_next(0, 200, 200));
-        assert!(topic_list_should_fetch_next(200, 1, 450));
-        assert!(!topic_list_should_fetch_next(400, 50, 450));
+    fn topic_infos_from_agent_list_response_parses_full_catalog() {
+        let topics: Vec<serde_json::Value> = (0..341)
+            .map(|i| serde_json::json!({ "name": format!("topic-{i}"), "partitions": 4, "messageType": "UNSPECIFIED" }))
+            .collect();
+        let response = serde_json::json!({ "topics": topics, "total": 341, "offset": 0, "limit": i32::MAX });
+        let parsed = topic_infos_from_agent_list_response(&response);
+        assert_eq!(parsed.len(), 341);
+        assert_eq!(agent_list_total(&response, parsed.len()), 341);
+        assert_eq!(parsed.first().map(|t| t.name.as_str()), Some("topic-0"));
+        assert_eq!(parsed.last().map(|t| t.name.as_str()), Some("topic-340"));
+        assert_eq!(parsed[0].message_type.as_deref(), Some("UNSPECIFIED"));
     }
 
     #[test]
-    fn topic_infos_from_agent_pages_merges_200_201_and_multi_page_totals() {
-        let page1: Vec<serde_json::Value> =
+    fn agent_list_total_detects_truncated_first_page() {
+        let page: Vec<serde_json::Value> =
             (0..200).map(|i| serde_json::json!({ "name": format!("topic-{i}"), "partitions": 4 })).collect();
-        let page2_201 = serde_json::json!({ "name": "topic-200", "partitions": 4 });
-        let response_201_first = serde_json::json!({ "topics": page1, "total": 201, "offset": 0, "limit": 200 });
-        let response_201_second =
-            serde_json::json!({ "topics": [page2_201], "total": 201, "offset": 200, "limit": 200 });
-        let merged_201 = topic_infos_from_agent_pages(&[response_201_first, response_201_second]);
-        assert_eq!(merged_201.len(), 201);
-        assert_eq!(merged_201.last().map(|t| t.name.as_str()), Some("topic-200"));
-
-        let page_a: Vec<serde_json::Value> =
-            (0..200).map(|i| serde_json::json!({ "name": format!("p-{i}"), "partitions": 1 })).collect();
-        let page_b: Vec<serde_json::Value> =
-            (200..400).map(|i| serde_json::json!({ "name": format!("p-{i}"), "partitions": 1 })).collect();
-        let page_c: Vec<serde_json::Value> =
-            (400..450).map(|i| serde_json::json!({ "name": format!("p-{i}"), "partitions": 1 })).collect();
-        let merged_450 = topic_infos_from_agent_pages(&[
-            serde_json::json!({ "topics": page_a, "total": 450 }),
-            serde_json::json!({ "topics": page_b, "total": 450 }),
-            serde_json::json!({ "topics": page_c, "total": 450 }),
-        ]);
-        assert_eq!(merged_450.len(), 450);
-        assert_eq!(merged_450.first().map(|t| t.name.as_str()), Some("p-0"));
-        assert_eq!(merged_450.last().map(|t| t.name.as_str()), Some("p-449"));
+        let response = serde_json::json!({ "topics": page, "total": 338, "offset": 0, "limit": 200 });
+        let parsed = topic_infos_from_agent_list_response(&response);
+        assert_eq!(parsed.len(), 200);
+        assert_eq!(agent_list_total(&response, parsed.len()), 338);
+        assert!((parsed.len() as u64) < agent_list_total(&response, parsed.len()));
     }
 
     #[test]
@@ -1102,4 +1096,18 @@ mod tests {
         assert_eq!(peeked.properties.get("tag").map(String::as_str), Some("cs-pt-dlq-test"));
         assert_eq!(peeked.publish_time.as_deref(), Some("1710000000000"));
     }
+
+    #[test]
+    fn cluster_info_from_agent_result_parses_acl_and_brokers() {
+        let result = serde_json::json!({
+            "ok": true,
+            "clusterId": "DefaultCluster",
+            "brokers": [{"brokerName": "broker-a"}],
+            "aclEnabled": false
+        });
+        let info = super::cluster_info_from_agent_result(&result);
+        assert_eq!(info.system_kind, MqSystemKind::RocketMq);
+        assert!(!info.capabilities.supports_permissions);
+        assert_eq!(info.extra.get("clusterId").and_then(|v| v.as_str()), Some("DefaultCluster"));
+    }
 }

crates/dbx-core/src/mq/mod.rs

@@ -56,6 +56,13 @@ struct CachedMqAdmin {
     adapter: Arc<dyn MessageQueueAdmin>,
 }
 
+/// Result of resolving an MQ admin adapter for a connection attempt.
+pub struct MqBuildResult {
+    pub adapter: Arc<dyn MessageQueueAdmin>,
+    /// True when an existing cached adapter was reused (reconnect fast path).
+    pub was_cached: bool,
+}
+
 impl MqAdminRegistry {
     pub fn new() -> Self {
         Self { instances: RwLock::new(HashMap::new()), build_locks: RwLock::new(HashMap::new()) }
@@ -63,7 +70,7 @@ impl MqAdminRegistry {
 
     /// Return the cached adapter for this connection, building it from the
     /// connection's `external_config` if not already present.
-    pub async fn get_or_build(&self, cfg: &ConnectionConfig) -> Result<Arc<dyn MessageQueueAdmin>, String> {
+    pub async fn get_or_build(&self, cfg: &ConnectionConfig) -> Result<MqBuildResult, String> {
         let mqc = MqAdminConfig::from_connection(cfg)?;
         self.get_or_build_config(&cfg.id, mqc, None).await
     }
@@ -73,13 +80,13 @@ impl MqAdminRegistry {
         connection_id: &str,
         mqc: MqAdminConfig,
         agent_launch: Option<AgentLaunchSpec>,
-    ) -> Result<Arc<dyn MessageQueueAdmin>, String> {
+    ) -> Result<MqBuildResult, String> {
         let fingerprint = adapter_fingerprint(&mqc, agent_launch.as_ref());
 
         // Fast path: return the cached adapter.
         if let Some(entry) = self.instances.read().await.get(connection_id) {
             if entry.fingerprint == fingerprint {
-                return Ok(entry.adapter.clone());
+                return Ok(MqBuildResult { adapter: entry.adapter.clone(), was_cached: true });
             }
         }
 
@@ -94,16 +101,19 @@ impl MqAdminRegistry {
         // Another task may have built it while we were waiting for the lock.
         if let Some(entry) = self.instances.read().await.get(connection_id) {
             if entry.fingerprint == fingerprint {
-                return Ok(entry.adapter.clone());
+                return Ok(MqBuildResult { adapter: entry.adapter.clone(), was_cached: true });
             }
         }
 
+        // Config changed — drop the stale adapter so its agent process is released.
+        self.instances.write().await.remove(connection_id);
+
         let adapter = build_adapter(mqc, agent_launch).await?;
         self.instances
             .write()
             .await
             .insert(connection_id.to_string(), CachedMqAdmin { fingerprint, adapter: adapter.clone() });
-        Ok(adapter)
+        Ok(MqBuildResult { adapter, was_cached: false })
     }
 
     /// Drop the cached adapter for a connection (called on disconnect).
@@ -112,6 +122,16 @@ impl MqAdminRegistry {
         self.build_locks.write().await.remove(connection_id);
     }
 
+    /// Whether an adapter is currently cached for this connection id.
+    pub async fn has_cached_connection(&self, connection_id: &str) -> bool {
+        self.instances.read().await.contains_key(connection_id)
+    }
+
+    /// Connection ids currently holding a cached MQ adapter (for cleanup assertions).
+    pub async fn cached_connection_ids(&self) -> Vec<String> {
+        self.instances.read().await.keys().cloned().collect()
+    }
+
     /// Build a fresh adapter without caching it — used for connection tests
     /// where we don't want to retain state.
     pub async fn build_transient(&self, cfg: &ConnectionConfig) -> Result<Arc<dyn MessageQueueAdmin>, String> {
@@ -128,6 +148,15 @@ impl MqAdminRegistry {
     }
 }
 
+/// Validate connectivity after resolving an MQ adapter. Skips an immediate probe when
+/// the adapter was just built and its connect path already verified the cluster.
+pub async fn validate_mq_adapter_after_build(build: &MqBuildResult) -> Result<(), String> {
+    if build.was_cached || !build.adapter.build_includes_connect_test() {
+        build.adapter.test_connection().await?;
+    }
+    Ok(())
+}
+
 fn adapter_fingerprint(mqc: &MqAdminConfig, agent_launch: Option<&AgentLaunchSpec>) -> u64 {
     let mut hasher = std::collections::hash_map::DefaultHasher::new();
     format!("{mqc:?}").hash(&mut hasher);

crates/dbx-core/src/mq/port.rs

@@ -18,6 +18,13 @@ pub trait MessageQueueAdmin: Send + Sync {
     /// Connectivity test; returns cluster/version info.
     async fn test_connection(&self) -> Result<MqClusterInfo, String>;
 
+    /// When true, the adapter's build path already validated connectivity (e.g.
+    /// RocketMQ agent `connect` RPC). Callers may skip an immediate follow-up
+    /// `test_connection` on first build, but should still test on cache reuse.
+    fn build_includes_connect_test(&self) -> bool {
+        false
+    }
+
     // ---- Tenants ----
     async fn list_tenants(&self) -> Result<Vec<TenantInfo>, String>;
     async fn get_tenant(&self, name: &str) -> Result<TenantInfo, String>;

crates/dbx-core/src/mq/service.rs

@@ -22,14 +22,14 @@ pub async fn mq_test_connection_core(state: &AppState, conn_id: &str) -> Result<
     let cfg = state.configs.read().await.get(conn_id).cloned().ok_or("Connection not found")?;
     let mqc = state.mq_admin_config_for_connection(conn_id, &cfg).await?;
     let agent_launch = resolve_mq_agent_launch_spec(&mqc, state);
-    let adapter = match state.mq_registry.get_or_build_config(conn_id, mqc, agent_launch).await {
-        Ok(adapter) => adapter,
+    let build = match state.mq_registry.get_or_build_config(conn_id, mqc, agent_launch).await {
+        Ok(build) => build,
         Err(err) => {
             state.mq_registry.drop_connection(conn_id).await;
             return Err(err);
         }
     };
-    match adapter.test_connection().await {
+    match build.adapter.test_connection().await {
         Ok(info) => Ok(info),
         Err(err) => {
             state.mq_registry.drop_connection(conn_id).await;
@@ -801,7 +801,7 @@ async fn get_adapter(
     let cfg = state.configs.read().await.get(conn_id).cloned().ok_or("Connection not found")?;
     let mqc = state.mq_admin_config_for_connection(conn_id, &cfg).await?;
     let agent_launch = resolve_mq_agent_launch_spec(&mqc, state);
-    state.mq_registry.get_or_build_config(conn_id, mqc, agent_launch).await
+    state.mq_registry.get_or_build_config(conn_id, mqc, agent_launch).await.map(|build| build.adapter)
 }
 
 /// Resolve the MQ agent launch spec for agent-backed systems (Kafka, RocketMQ, RabbitMQ).

crates/dbx-web/src/routes/connection.rs

@@ -181,6 +181,10 @@ async fn run_temporary_connection_test(
     };
 
     app.remove_connection_pools(&temp_id).await;
+    // Pool drain intentionally keeps durable MQ adapters for reconnect reuse; temporary
+    // probes must still release any registry entry if a cached path was used.
+    #[cfg(feature = "mq-admin")]
+    app.mq_registry.drop_connection(&temp_id).await;
     app.reset_connection_transport_for_config(&temp_id, &config).await;
     app.configs.write().await.remove(&temp_id);
 
@@ -669,6 +673,30 @@ mod tests {
         let _ = std::fs::remove_dir_all(dir);
     }
 
+    #[cfg(feature = "mq-admin")]
+    #[tokio::test]
+    async fn mq_connection_test_does_not_retain_temporary_adapter() {
+        let (state, dir) = test_web_state().await;
+        let admin_url = spawn_pulsar_clusters_server().await;
+        let config = mq_config("pulsar-probe", &admin_url);
+
+        let result = test_connection(State(state.clone()), Json(ConnectRequest { config, client_attempt: None }))
+            .await
+            .unwrap_or_else(|error| panic!("{}", error.message));
+        assert_eq!(result.0, "Connection successful");
+
+        assert!(state.app.configs.read().await.keys().all(|key| !key.starts_with("__test_")));
+        assert!(state.app.connections.read().await.keys().all(|key| !key.starts_with("__test_")));
+        let cached = state.app.mq_registry.cached_connection_ids().await;
+        assert!(
+            cached.iter().all(|id| !id.starts_with("__test_")),
+            "temporary MQ connection tests must not retain registry adapters: {cached:?}"
+        );
+        assert!(!state.app.mq_registry.has_cached_connection("pulsar-probe").await);
+
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     #[tokio::test]
     async fn invalid_persistent_sqlite_attachments_do_not_replace_live_web_state() {
         let (state, dir) = test_web_state().await;
@@ -898,7 +926,7 @@ mod tests {
         let initial = mq_config("mq-conn", "http://127.0.0.1:8080");
         state.app.configs.write().await.insert(initial.id.clone(), initial.clone());
         state.app.connections.write().await.insert(initial.id.clone(), PoolKind::MessageQueue);
-        let first = state.app.mq_registry.get_or_build(&initial).await.unwrap();
+        let first = state.app.mq_registry.get_or_build(&initial).await.unwrap().adapter;
 
         let updated = mq_config("mq-conn", "http://127.0.0.1:8081");
         let result =
@@ -918,7 +946,7 @@ mod tests {
             .map(str::to_string);
         assert_eq!(cached_admin_url.as_deref(), Some("http://127.0.0.1:8081"));
 
-        let second = state.app.mq_registry.get_or_build(&updated).await.unwrap();
+        let second = state.app.mq_registry.get_or_build(&updated).await.unwrap().adapter;
         assert!(!Arc::ptr_eq(&first, &second));
         assert!(!state.app.connections.read().await.contains_key(&initial.id));
 
@@ -932,7 +960,7 @@ mod tests {
         let initial = mq_config("mq-conn", "http://127.0.0.1:8080");
         state.app.configs.write().await.insert(initial.id.clone(), initial.clone());
         state.app.connections.write().await.insert(initial.id.clone(), PoolKind::MessageQueue);
-        let first = state.app.mq_registry.get_or_build(&initial).await.unwrap();
+        let first = state.app.mq_registry.get_or_build(&initial).await.unwrap().adapter;
 
         let updated = mq_config("mq-conn", &spawn_pulsar_clusters_server().await);
         let result =
@@ -940,7 +968,7 @@ mod tests {
                 .await;
         assert!(result.is_ok());
 
-        let second = state.app.mq_registry.get_or_build(&updated).await.unwrap();
+        let second = state.app.mq_registry.get_or_build(&updated).await.unwrap().adapter;
         assert!(!Arc::ptr_eq(&first, &second));
 
         let _ = std::fs::remove_dir_all(dir);
@@ -982,7 +1010,7 @@ mod tests {
             configs.insert(kept.id.clone(), kept.clone());
             configs.insert(removed.id.clone(), removed.clone());
         }
-        let stale = state.app.mq_registry.get_or_build(&removed).await.unwrap();
+        let stale = state.app.mq_registry.get_or_build(&removed).await.unwrap().adapter;
 
         let result =
             save_connections(State(state.clone()), Json(SaveConnectionsRequest { configs: vec![kept.clone()] })).await;
@@ -993,7 +1021,7 @@ mod tests {
         assert!(!configs.contains_key("removed-mq"));
         drop(configs);
 
-        let rebuilt = state.app.mq_registry.get_or_build(&removed).await.unwrap();
+        let rebuilt = state.app.mq_registry.get_or_build(&removed).await.unwrap().adapter;
         assert!(!Arc::ptr_eq(&stale, &rebuilt));
 
         let _ = std::fs::remove_dir_all(dir);
@@ -1120,7 +1148,7 @@ mod tests {
         let config = mq_config("mq-conn", "http://127.0.0.1:8080");
         state.app.configs.write().await.insert(config.id.clone(), config.clone());
         state.app.connections.write().await.insert(config.id.clone(), PoolKind::MessageQueue);
-        let first = state.app.mq_registry.get_or_build(&config).await.unwrap();
+        let first = state.app.mq_registry.get_or_build(&config).await.unwrap().adapter;
 
         let result = disconnect_db(
             State(state.clone()),
@@ -1130,7 +1158,7 @@ mod tests {
         assert!(result.is_ok());
 
         assert!(!state.app.connections.read().await.contains_key(&config.id));
-        let second = state.app.mq_registry.get_or_build(&config).await.unwrap();
+        let second = state.app.mq_registry.get_or_build(&config).await.unwrap().adapter;
         assert!(!Arc::ptr_eq(&first, &second));
 
         let _ = std::fs::remove_dir_all(dir);

scripts/check-mq-integration.mjs

@@ -20,7 +20,7 @@ assertMatches("crates/dbx-web/Cargo.toml", /\[features\][\s\S]*default\s*=\s*\[[
 
 assertIncludes("apps/desktop/src/components/layout/ContentArea.vue", "MqAdminConsole", "Main content area must import and render the MQ admin console.");
 assertIncludes("apps/desktop/src/components/layout/ContentArea.vue", "activeTab.mode === 'mq'", "Main content area must have an mq mode render branch.");
-assertIncludes("apps/desktop/src/components/sidebar/TreeItem.vue", '"mq"', "Sidebar connection handling must be aware of MQ connections.");
+assertIncludes("apps/desktop/src/components/sidebar/TreeItem.vue", "mq-tenant", "Sidebar tree items must render MQ tenant nodes.");
 assertIncludes("apps/desktop/src/stores/queryStore.ts", "openMqAdmin", "Query store must expose an MQ admin tab opener.");
 
 const manifest = JSON.parse(read("crates/dbx-core/assets/database-drivers.manifest.json"));
@@ -80,6 +80,16 @@ for (const panel of ["PoliciesPanel.vue", "PermissionsPanel.vue", "RawApiPanel.v
 for (const panel of ["TenantsPanel.vue", "NamespacesPanel.vue", "TopicsPanel.vue", "SubscriptionsPanel.vue"]) {
   assertIncludes(`apps/desktop/src/components/mq/${panel}`, "readOnly", `${panel} must disable mutating actions in read-only mode.`);
 }
+assertIncludes(
+  "apps/desktop/src/components/mq/TopicsPanel.vue",
+  "topics-table-hscroll",
+  "TopicsPanel must keep a shared horizontal scroller so wide RocketMQ action columns stay reachable.",
+);
+assertIncludes(
+  "apps/desktop/src/components/mq/TopicsPanel.vue",
+  "overflow-x: auto",
+  "TopicsPanel horizontal scroller must allow overflow-x.",
+);
 for (const panel of ["ExchangesPanel.vue", "SendMessagePanel.vue", "rabbitmq/RabbitMqClientsPanel.vue", "ProducerConsumerPanel.vue"]) {
   assertIncludes(`apps/desktop/src/components/mq/${panel}`, "readOnly", `${panel} must disable mutating actions in read-only mode.`);
 }

src-tauri/src/commands/connection.rs

@@ -485,7 +485,7 @@ mod tests {
         let initial = mq_config("mq-conn", "http://127.0.0.1:8080");
         state.configs.write().await.insert(initial.id.clone(), initial.clone());
         state.connections.write().await.insert(initial.id.clone(), PoolKind::MessageQueue);
-        let first = state.mq_registry.get_or_build(&initial).await.unwrap();
+        let first = state.mq_registry.get_or_build(&initial).await.unwrap().adapter;
 
         let updated = mq_config("mq-conn", "http://127.0.0.1:8081");
         save_connection_configs(&state, std::slice::from_ref(&updated)).await.unwrap();
@@ -501,7 +501,7 @@ mod tests {
             .map(str::to_string);
         assert_eq!(cached_admin_url.as_deref(), Some("http://127.0.0.1:8081"));
 
-        let second = state.mq_registry.get_or_build(&updated).await.unwrap();
+        let second = state.mq_registry.get_or_build(&updated).await.unwrap().adapter;
         assert!(!std::sync::Arc::ptr_eq(&first, &second));
         assert!(!state.connections.read().await.contains_key(&initial.id));
 
@@ -553,7 +553,7 @@ mod tests {
             configs.insert(kept.id.clone(), kept.clone());
             configs.insert(removed.id.clone(), removed.clone());
         }
-        let stale = state.mq_registry.get_or_build(&removed).await.unwrap();
+        let stale = state.mq_registry.get_or_build(&removed).await.unwrap().adapter;
 
         save_connection_configs(&state, std::slice::from_ref(&kept)).await.unwrap();
 
@@ -562,7 +562,7 @@ mod tests {
         assert!(!configs.contains_key("removed-mq"));
         drop(configs);
 
-        let rebuilt = state.mq_registry.get_or_build(&removed).await.unwrap();
+        let rebuilt = state.mq_registry.get_or_build(&removed).await.unwrap().adapter;
         assert!(!std::sync::Arc::ptr_eq(&stale, &rebuilt));
 
         let _ = std::fs::remove_dir_all(dir);
@@ -1085,19 +1085,12 @@ async fn test_connection_with_info_inner(
             }
             #[cfg(feature = "mq-admin")]
             DatabaseType::MessageQueue => {
+                // Probe with a transient adapter so Test Connection never retains/replaces
+                // a live cached MQ agent for this connection id (same pattern as Nacos).
                 let mqc = state.mq_admin_config_for_connection(connection_id, &config).await?;
                 let agent_launch = dbx_core::mq::service::resolve_mq_agent_launch_spec(&mqc, state);
-                let adapter = match state.mq_registry.get_or_build_config(connection_id, mqc, agent_launch).await {
-                    Ok(adapter) => adapter,
-                    Err(err) => {
-                        state.mq_registry.drop_connection(connection_id).await;
-                        return Err(err);
-                    }
-                };
-                if let Err(err) = adapter.test_connection().await {
-                    state.mq_registry.drop_connection(connection_id).await;
-                    return Err(err);
-                }
+                let adapter = state.mq_registry.build_transient_config(mqc, agent_launch).await?;
+                adapter.test_connection().await?;
                 Ok("Connection successful".to_string())
             }
             #[cfg(not(feature = "mq-admin"))]
@@ -1420,8 +1413,8 @@ pub async fn connect_db(
         DatabaseType::MessageQueue => {
             let mqc = state.mq_admin_config_for_connection(&id, &config).await?;
             let agent_launch = dbx_core::mq::service::resolve_mq_agent_launch_spec(&mqc, &state);
-            let adapter = match state.mq_registry.get_or_build_config(&id, mqc, agent_launch).await {
-                Ok(adapter) => adapter,
+            let build = match state.mq_registry.get_or_build_config(&id, mqc, agent_launch).await {
+                Ok(build) => build,
                 Err(err) => {
                     state.mq_registry.drop_connection(&id).await;
                     return Err(err);
@@ -1431,7 +1424,7 @@ pub async fn connect_db(
                 state.mq_registry.drop_connection(&id).await;
                 return Err(err);
             }
-            if let Err(err) = adapter.test_connection().await {
+            if let Err(err) = dbx_core::mq::validate_mq_adapter_after_build(&build).await {
                 state.mq_registry.drop_connection(&id).await;
                 return Err(err);
             }