diff --git a/.gitattributes b/.gitattributes index c89d144f1..4939cbd20 100644 --- a/.gitattributes +++ b/.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. diff --git a/agents/drivers/rocketmq/src/main/java/com/dbx/agent/rocketmq/RocketMqAgent.java b/agents/drivers/rocketmq/src/main/java/com/dbx/agent/rocketmq/RocketMqAgent.java index c39f120ae..5236ba7f9 100644 --- a/agents/drivers/rocketmq/src/main/java/com/dbx/agent/rocketmq/RocketMqAgent.java +++ b/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> brokers = brokerNodes(clusterInfo); - boolean aclEnabled = probeAclSupport(probe); - - Map 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 buildClusterTestResult( + ClusterInfo clusterInfo, + DefaultMQAdminExt admin, + JsonObject conn + ) throws Exception { + String clusterName = resolveClusterName(clusterInfo, conn); + List> brokers = brokerNodes(clusterInfo); + boolean aclEnabled = probeAclSupport(admin, clusterInfo); + + Map 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 brokerSystemTopics = collectBrokerSystemTopics(admin); + Set brokerNames = collectBrokerNames(clusterInfo); + String cluster = resolveClusterName(clusterInfo, conn); + BrokerTopicConfigSnapshot brokerSnapshot = collectBrokerTopicConfigSnapshot(admin, conn, clusterInfo); + Map brokerTopics = brokerSnapshot.topics(); + Map> topicAttributes = topicAttributesFromConfigs(brokerTopics); + + Set 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 topicNames = topicCatalogNames(brokerSnapshot, nameserverTopics); + + List> 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> page = paginate(topics, offset, limit); + Map 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 topics; + private final boolean complete; + + private BrokerTopicConfigSnapshot(Map topics, boolean complete) { + this.topics = topics; + this.complete = complete; } - TopicList topicList = fetchTopicList(admin, connectionObject(params)); - Set brokerSystemTopics = collectBrokerSystemTopics(admin); - Set brokerNames = collectBrokerNames(admin); - String cluster = clusterName(params, admin); - JsonObject conn = connectionObject(params); - Map brokerTopics = collectBrokerTopicConfigs(admin, conn); - Map> topicAttributes = topicAttributesFromConfigs(brokerTopics); - // Prefer broker topic configs (Dashboard ground truth); nameserver routes can outlive broker deletion. - Set topicNames = brokerTopics.isEmpty() - ? new TreeSet<>(topicList.getTopicList()) - : brokerTopics.keySet(); + Map topics() { + return topics; + } + + boolean complete() { + return complete; + } + } + + static Set topicCatalogNames( + BrokerTopicConfigSnapshot brokerSnapshot, Set nameserverTopics) { + if (brokerSnapshot.complete() && !brokerSnapshot.topics().isEmpty()) { + return new TreeSet<>(brokerSnapshot.topics().keySet()); + } + Set 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> buildTopicCatalogRows( + Set topicNames, + Map brokerTopics, + Map> topicAttributes, + Set brokerSystemTopics, + Set 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> topics = new ArrayList<>(); + List> 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 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 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> page = paginate(topics, offset, limit); - Map 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> 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> tasks = new ArrayList<>(rows.size()); + for (Map row : rows) { + String topic = String.valueOf(row.get("name")); + tasks.add(() -> { + try { + return partitionsFromRoute(lookup.load(topic)); + } catch (Exception ignored) { + return 1; + } + }); + } + List> 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 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 resolveMasterBrokerAddrs( DefaultMQAdminExt admin, JsonObject conn, String brokerNameFilter) throws Exception { - ClusterInfo clusterInfo = admin.examineBrokerClusterInfo(); - LinkedHashSet 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 resolveMasterBrokerAddrs( + DefaultMQAdminExt admin, JsonObject conn, String brokerNameFilter, ClusterInfo clusterInfo) + throws Exception { + LinkedHashSet 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 masterBrokerAddrsFromClusterInfo( + ClusterInfo clusterInfo, JsonObject conn, String brokerNameFilter) { + LinkedHashSet 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 collectBrokerNames(DefaultMQAdminExt admin) { - Set 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 collectBrokerNames(ClusterInfo clusterInfo) { + Set 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 collectBrokerTopicConfigs(DefaultMQAdminExt admin, JsonObject conn) { - Map 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 brokerAddrs = resolveMasterBrokerAddrs(admin, conn, null, clusterInfo); + return collectBrokerTopicConfigs( + brokerAddrs, + brokerAddr -> admin.getAllTopicConfig(brokerAddr, DEFAULT_REQUEST_TIMEOUT_MS) + ); + } + + static BrokerTopicConfigSnapshot collectBrokerTopicConfigs( + List brokerAddrs, BrokerTopicConfigLookup lookup) { + Map 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 List paginate(List 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; } diff --git a/agents/drivers/rocketmq/src/test/java/com/dbx/agent/rocketmq/RocketMqAgentTest.java b/agents/drivers/rocketmq/src/test/java/com/dbx/agent/rocketmq/RocketMqAgentTest.java index e0531cb32..6140b55d3 100644 --- a/agents/drivers/rocketmq/src/test/java/com/dbx/agent/rocketmq/RocketMqAgentTest.java +++ b/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 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 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> rows = IntStream.range(0, 4) + .mapToObj(index -> { + Map 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 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 row : rows) { + assertEquals(8, row.get("partitions")); + } + } + + @Test + void enrichTopicPartitionsReturnsDefaultsWhenBudgetExpires() { + List> rows = IntStream.range(0, 4) + .mapToObj(index -> { + Map 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 row : rows) { + assertEquals(1, row.get("partitions")); + } + } + + @Test + void buildTopicCatalogRowsFallbackIssuesAtMostOneRouteRpcPerTopic() { + Set topicNames = new TreeSet<>(List.of("Alpha", "Beta", "Gamma")); + Map routeCalls = new ConcurrentHashMap<>(); + AtomicInteger totalRouteCalls = new AtomicInteger(); + + List> 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 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 brokerTopics = Map.of("Orders", config); + AtomicInteger routeCalls = new AtomicInteger(); + + List> 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)); + } } diff --git a/apps/desktop/src/components/mq/MessageQueryPanel.vue b/apps/desktop/src/components/mq/MessageQueryPanel.vue index c23eca46f..1cb3fedc1 100644 --- a/apps/desktop/src/components/mq/MessageQueryPanel.vue +++ b/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, () => { - +