[ISSUES #281 #274 #285 #287] Speeds up topic and consumer queries, adds caching, and fixes delay/dead-letter topic mix-up (#286)

* fix: Resolved issue of query failure under a large number of topics and consumers #281

* fix: Expand the message ID query time range to avoid query failure

* fix: Remove duplicates from topic queries, increase system topic recognition #287

Author: Crazylychee

src/main/java/org/apache/rocketmq/dashboard/controller/ConsumerController.java

@@ -51,6 +51,13 @@ public Object list(@RequestParam(value = "skipSysGroup", required = false) boole
         return consumerService.queryGroupList(skipSysGroup, address);
     }
 
+    @RequestMapping(value = "/group.refresh")
+    @ResponseBody
+    public Object refresh(String address,
+        String consumerGroup) {
+        return consumerService.refreshGroup(address, consumerGroup);
+    }
+
     @RequestMapping(value = "/group.query")
     @ResponseBody
     public Object groupQuery(@RequestParam String consumerGroup, String address) {

src/main/java/org/apache/rocketmq/dashboard/controller/TopicController.java

@@ -56,6 +56,12 @@ public Object list(@RequestParam(value = "skipSysProcess", required = false) boo
         return topicService.fetchAllTopicList(skipSysProcess, skipRetryAndDlq);
     }
 
+    @RequestMapping(value = "/refresh", method = {RequestMethod.POST})
+    @ResponseBody
+    public Object refresh() {
+        return topicService.refreshTopicList();
+    }
+
     @RequestMapping(value = "/list.queryTopicType", method = RequestMethod.GET)
     @ResponseBody
     public Object listTopicType() {

src/main/java/org/apache/rocketmq/dashboard/model/GroupConsumeInfo.java

@@ -19,6 +19,7 @@
 import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
 import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
 
+import java.util.Date;
 import java.util.List;
 
 public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
@@ -31,6 +32,7 @@ public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
     private int consumeTps;
     private long diffTotal = -1;
     private String subGroupType = "NORMAL";
+    private Date updateTime;
 
 
     public String getGroup() {
@@ -112,4 +114,12 @@ public String getVersion() {
     public void setVersion(String version) {
         this.version = version;
     }
+
+    public Date getUpdateTime() {
+        return updateTime;
+    }
+
+    public void setUpdateTime(Date updateTime) {
+        this.updateTime = updateTime;
+    }
 }

src/main/java/org/apache/rocketmq/dashboard/service/ClusterInfoService.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.dashboard.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.Resource;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+@Slf4j
+@Service
+public class ClusterInfoService {
+
+    @Resource
+    private MQAdminExt mqAdminExt;
+
+    @Value("${rocketmq.cluster.cache.expire:60000}")
+    private long cacheExpireMs;
+
+
+    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+    private final AtomicReference<ClusterInfo> cachedRef = new AtomicReference<>();
+
+
+    @PostConstruct
+    public void init() {
+        scheduler.scheduleAtFixedRate(this::refresh,
+                0, cacheExpireMs / 2, TimeUnit.MILLISECONDS);
+    }
+
+    public ClusterInfo get() {
+        ClusterInfo info = cachedRef.get();
+        return info != null ? info : refresh();
+    }
+
+    public synchronized ClusterInfo refresh() {
+        try {
+            ClusterInfo fresh = mqAdminExt.examineBrokerClusterInfo();
+            cachedRef.set(fresh);
+            return fresh;
+        } catch (Exception e) {
+            log.warn("Refresh cluster info failed", e);
+            ClusterInfo old = cachedRef.get();
+            if (old != null) {
+                return old;
+            }
+            throw new IllegalStateException("No cluster info available", e);
+        }
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/service/ConsumerService.java

@@ -17,14 +17,14 @@
 
 package org.apache.rocketmq.dashboard.service;
 
+import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
+import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
+import org.apache.rocketmq.dashboard.model.request.ResetOffsetRequest;
 import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
 import org.apache.rocketmq.dashboard.model.ConsumerGroupRollBackStat;
 import org.apache.rocketmq.dashboard.model.GroupConsumeInfo;
 import org.apache.rocketmq.dashboard.model.TopicConsumerInfo;
-import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
-import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
-import org.apache.rocketmq.dashboard.model.request.ResetOffsetRequest;
 
 import java.util.List;
 import java.util.Map;
@@ -55,4 +55,6 @@ public interface ConsumerService {
     ConsumerConnection getConsumerConnection(String consumerGroup, String address);
 
     ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack);
+
+    Object refreshGroup(String address, String consumerGroup);
 }

src/main/java/org/apache/rocketmq/dashboard/service/TopicService.java

@@ -54,4 +54,5 @@ public interface TopicService {
 
     SendResult sendTopicMessageRequest(SendTopicMessageRequest sendTopicMessageRequest);
 
+    boolean refreshTopicList();
 }

src/main/java/org/apache/rocketmq/dashboard/service/client/MQAdminExtImpl.java

@@ -17,11 +17,15 @@
 package org.apache.rocketmq.dashboard.service.client;
 
 import com.google.common.base.Throwables;
+
 import java.io.UnsupportedEncodingException;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
 import org.apache.rocketmq.client.QueryResult;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
@@ -87,9 +91,15 @@
 public class MQAdminExtImpl implements MQAdminExt {
     private Logger logger = LoggerFactory.getLogger(MQAdminExtImpl.class);
 
-    public MQAdminExtImpl() {
+    private static final ConcurrentMap<String, TopicConfigSerializeWrapper> TOPIC_CONFIG_CACHE = new ConcurrentHashMap<>();
+
+    public MQAdminExtImpl() {}
+
+    public static void clearTopicConfigCache() {
+        TOPIC_CONFIG_CACHE.clear();
     }
 
+
     @Override
     public void updateBrokerConfig(String brokerAddr, Properties properties)
         throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
@@ -145,7 +155,7 @@ public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, Strin
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null);
         RemotingCommand response = null;
         try {
-            response = remotingClient.invokeSync(addr, request, 3000);
+            response = remotingClient.invokeSync(addr, request, 8000);
         }
         catch (Exception err) {
             Throwables.throwIfUnchecked(err);
@@ -164,19 +174,27 @@ public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, Strin
 
     @Override
     public TopicConfig examineTopicConfig(String addr, String topic) throws MQBrokerException {
+        TopicConfigSerializeWrapper cachedWrapper = TOPIC_CONFIG_CACHE.get(addr);
+
+        if (cachedWrapper != null && cachedWrapper.getTopicConfigTable().containsKey(topic)) {
+            return cachedWrapper.getTopicConfigTable().get(topic);
+        }
+
         RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
         RemotingCommand response = null;
         try {
             response = remotingClient.invokeSync(addr, request, 3000);
-        }
-        catch (Exception err) {
+        } catch (Exception err) {
             Throwables.throwIfUnchecked(err);
             throw new RuntimeException(err);
         }
         switch (response.getCode()) {
             case ResponseCode.SUCCESS: {
-                TopicConfigSerializeWrapper topicConfigSerializeWrapper = decode(response.getBody(), TopicConfigSerializeWrapper.class);
+                TopicConfigSerializeWrapper topicConfigSerializeWrapper =
+                        decode(response.getBody(), TopicConfigSerializeWrapper.class);
+
+                TOPIC_CONFIG_CACHE.put(addr, topicConfigSerializeWrapper);
                 return topicConfigSerializeWrapper.getTopicConfigTable().get(topic);
             }
             default:
@@ -468,14 +486,14 @@ public MessageExt viewMessage(String topic,
         Set<String> clusterList = MQAdminInstance.threadLocalMQAdminExt().getTopicClusterList(topic);
         if (clusterList == null || clusterList.isEmpty()) {
             QueryResult qr = Reflect.on(mqAdminImpl).call("queryMessage", "", topic, msgId, 32,
-                    MessageClientIDSetter.getNearlyTimeFromID(msgId).getTime(), Long.MAX_VALUE, true).get();
+                    0L, Long.MAX_VALUE, true).get();
             if (qr != null && qr.getMessageList() != null && !qr.getMessageList().isEmpty()) {
                 return qr.getMessageList().get(0);
             }
         } else {
             for (String name : clusterList) {
                 QueryResult qr = Reflect.on(mqAdminImpl).call("queryMessage", name, topic, msgId, 32,
-                        MessageClientIDSetter.getNearlyTimeFromID(msgId).getTime(), Long.MAX_VALUE, true).get();
+                        0L, Long.MAX_VALUE, true).get();
                 if (qr != null && qr.getMessageList() != null && !qr.getMessageList().isEmpty()) {
                     return qr.getMessageList().get(0);
                 }