Merge 0441a1deee into e57d423268

---
updated-dependencies:
- dependency-name: org.yaml:snakeyaml
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.gitignore

@@ -5,3 +5,4 @@
 .project
 .factorypath
 .settings/
+.vscode

docs/1_0_0/UserGuide_CN.md

@@ -63,6 +63,18 @@
 * 根据消息主题和消息Id进行消息的查询
 * 消息详情可以展示这条消息的详细信息，查看消息对应到具体消费组的消费情况（如果异常，可以查看具体的异常信息）。可以向指定的消费组重发消息。
 
+## RocketMQ-V5.0 仪表盘
+* 版本切换
+  * RocketMQ右上角可切换不同版本，用户可以自主选择 RocketMQ-5.x 或 RocketMQ-4.x 版本
+* 主题页面 
+  * 支持延迟/顺序/事务消息的筛选
+  * 支持延迟/顺序/事物/普通等多种消息类型主题的新增与更新
+* 消费页面
+  * 支持顺序消费类型订阅组的过滤
+  * 提供顺序消费类型订阅组的新增与更新，如果需要开启顺序消费，FIFO类型的订阅组一定需要打开consumeOrderlyEnable选项
+* 代理页面（RocketMQ 5.0新增） 
+  * 支持代理节点的新增与查询
+  * 支持代理节点地址配置：在application.yml中可对proxyAddr和proxyAddrs属性进行预配置
 
 ## HTTPS 方式访问Dashboard
 * HTTPS功能实际上是使用SpringBoot提供的配置功能即可完成，首先，需要有一个SSL KeyStore来存放服务端证书，可以使用本工程所提供的测试密钥库:

docs/1_0_0/UserGuide_EN.md

@@ -64,6 +64,18 @@
 * look over this message's detail info.you can see the message's consume state(each group has one line),show the exception message if has exception.
 you can send this message to the group you selected
 
+## RocketMQ-V5.0 dashboard
+* Version switching
+  * RocketMQ can switch between different versions in the upper right corner, and users can freely choose between RocketMQ-5.X or RocketMQ-4.X versions
+* Theme page
+  * Support filtering of delayed/sequential/transaction messages
+  * Support the addition and update of multiple message types such as delay, sequence, object, and ordinary themes
+* Consumption page
+  * Support filtering of subscription groups for fifo consumption types
+  * Provide the addition and update of subscription groups for sequential consumption types. If fifo consumption needs to be enabled, FIFO type subscription groups must have the consumeOrderlyEnable option enabled
+* Proxy page (Added in RocketMQ 5.0)
+  * Support for adding and querying proxy nodes
+  * Support proxy node address configuration: ProxyAddr and proxyAddrs properties can be pre configured in application.yml
 
 ## Access Dashboard with HTTPS
 * SpringBoot itself has provided the SSL configuration. You can use the project test Keystore:resources/rmqcngkeystore.jks. The store is generated with the following unix keytool commands:

frontend/package.json

@@ -9,7 +9,7 @@
     "react": "^17.0.2",
     "react-dom": "^17.0.2",
     "react-json-view": "^1.21.3",
-    "react-scripts": "4.0.3",
+    "react-scripts": "5.0.1",
     "web-vitals": "^1.0.1"
   },
   "proxy": "http://localhost:8080",

pom.xml

@@ -28,7 +28,7 @@
     <groupId>org.apache.rocketmq</groupId>
     <artifactId>rocketmq-dashboard</artifactId>
     <packaging>jar</packaging>
-    <version>1.0.1-SNAPSHOT</version>
+    <version>2.0.1-SNAPSHOT</version>
     <name>rocketmq-dashboard</name>
 
     <scm>
@@ -91,10 +91,10 @@
         <commons-io.version>2.4</commons-io.version>
         <commons-cli.version>1.2</commons-cli.version>
         <commons-collections.version>3.2.2</commons-collections.version>
-        <rocketmq.version>4.9.3</rocketmq.version>
+        <rocketmq.version>5.1.0</rocketmq.version>
         <surefire.version>2.19.1</surefire.version>
         <aspectj.version>1.9.6</aspectj.version>
-        <lombok.version>1.18.12</lombok.version>
+        <lombok.version>1.18.22</lombok.version>
         <main.basedir>${basedir}/../..</main.basedir>
         <docker.image.prefix>apacherocketmq</docker.image.prefix>
         <spring.boot.version>2.6.0</spring.boot.version>
@@ -104,7 +104,7 @@
         <easyexcel.version>2.2.10</easyexcel.version>
         <asm.version>4.2</asm.version>
         <junit.version>4.12</junit.version>
-        <snakeyaml.version>1.30</snakeyaml.version>
+        <snakeyaml.version>1.32</snakeyaml.version>
         <cglib.version>2.2.2</cglib.version>
         <joor.version>0.9.6</joor.version>
         <bcpkix-jdk15on.version>1.68</bcpkix-jdk15on.version>
@@ -167,6 +167,7 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-namesrv</artifactId>
             <version>${rocketmq.version}</version>
+            <scope>test</scope>
             <exclusions>
                 <exclusion>
                     <groupId>org.slf4j</groupId>
@@ -186,6 +187,7 @@
             <groupId>org.apache.rocketmq</groupId>
             <artifactId>rocketmq-broker</artifactId>
             <version>${rocketmq.version}</version>
+            <scope>test</scope>
             <exclusions>
                 <exclusion>
                     <groupId>org.slf4j</groupId>
@@ -459,7 +461,7 @@
                         <configuration>
                             <target>
                                 <copy todir="${project.build.directory}/classes/public">
-                                    <fileset dir="${project.basedir}/frontend/build"/>
+                                    <fileset dir="${project.basedir}/frontend/build" />
                                 </copy>
                             </target>
                         </configuration>

src/main/java/org/apache/rocketmq/dashboard/admin/MQAdminPooledObjectFactory.java

@@ -21,7 +21,7 @@ import org.apache.commons.collections.MapUtils;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 
 @Slf4j

src/main/java/org/apache/rocketmq/dashboard/config/RMQConfigure.java

@@ -43,6 +43,8 @@ public class RMQConfigure {
     //use rocketmq.namesrv.addr first,if it is empty,than use system proerty or system env
     private volatile String namesrvAddr = System.getProperty(MixAll.NAMESRV_ADDR_PROPERTY, System.getenv(MixAll.NAMESRV_ADDR_ENV));
 
+    private volatile String proxyAddr;
+
     private volatile String isVIPChannel = System.getProperty(SEND_MESSAGE_WITH_VIP_CHANNEL_PROPERTY, "true");
 
 
@@ -62,6 +64,8 @@ public class RMQConfigure {
 
     private List<String> namesrvAddrs = new ArrayList<>();
 
+    private List<String> proxyAddrs = new ArrayList<>();
+
     public String getAccessKey() {
         return accessKey;
     }
@@ -86,6 +90,25 @@ public class RMQConfigure {
         return namesrvAddrs;
     }
 
+    public List<String> getProxyAddrs() {
+        return this.proxyAddrs;
+    }
+
+    public void setProxyAddrs(List<String> proxyAddrs) {
+        this.proxyAddrs = proxyAddrs;
+        if (CollectionUtils.isNotEmpty(proxyAddrs)) {
+            this.setProxyAddr(proxyAddrs.get(0));
+        }
+    }
+
+    public String getProxyAddr() {
+        return proxyAddr;
+    }
+
+    public void setProxyAddr(String proxyAddr) {
+        this.proxyAddr = proxyAddr;
+    }
+
     public void setNamesrvAddrs(List<String> namesrvAddrs) {
         this.namesrvAddrs = namesrvAddrs;
         if (CollectionUtils.isNotEmpty(namesrvAddrs)) {

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

@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.controller;
 import com.google.common.base.Preconditions;
 import javax.annotation.Resource;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.dashboard.model.ConnectionInfo;
 import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
 import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
@@ -47,14 +47,14 @@ public class ConsumerController {
 
     @RequestMapping(value = "/groupList.query")
     @ResponseBody
-    public Object list(@RequestParam(value = "skipSysGroup", required = false) boolean skipSysGroup) {
-        return consumerService.queryGroupList(skipSysGroup);
+    public Object list(@RequestParam(value = "skipSysGroup", required = false) boolean skipSysGroup, String address) {
+        return consumerService.queryGroupList(skipSysGroup, address);
     }
 
     @RequestMapping(value = "/group.query")
     @ResponseBody
-    public Object groupQuery(@RequestParam String consumerGroup) {
-        return consumerService.queryGroup(consumerGroup);
+    public Object groupQuery(@RequestParam String consumerGroup, String address) {
+        return consumerService.queryGroup(consumerGroup, address);
     }
 
     @RequestMapping(value = "/resetOffset.do", method = {RequestMethod.POST})
@@ -99,14 +99,14 @@ public class ConsumerController {
 
     @RequestMapping(value = "/queryTopicByConsumer.query")
     @ResponseBody
-    public Object queryConsumerByTopic(@RequestParam String consumerGroup) {
-        return consumerService.queryConsumeStatsListByGroupName(consumerGroup);
+    public Object queryConsumerByTopic(@RequestParam String consumerGroup, String address) {
+        return consumerService.queryConsumeStatsListByGroupName(consumerGroup, address);
     }
 
     @RequestMapping(value = "/consumerConnection.query")
     @ResponseBody
-    public Object consumerConnection(@RequestParam(required = false) String consumerGroup) {
-        ConsumerConnection consumerConnection = consumerService.getConsumerConnection(consumerGroup);
+    public Object consumerConnection(@RequestParam(required = false) String consumerGroup, String address) {
+        ConsumerConnection consumerConnection = consumerService.getConsumerConnection(consumerGroup, address);
         consumerConnection.setConnectionSet(ConnectionInfo.buildConnectionInfoHashSet(consumerConnection.getConnectionSet()));
         return consumerConnection;
     }

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

@@ -18,7 +18,7 @@ package org.apache.rocketmq.dashboard.controller;
 
 import com.google.common.collect.Maps;
 import org.apache.rocketmq.common.Pair;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 import org.apache.rocketmq.dashboard.model.MessagePage;
 import org.apache.rocketmq.dashboard.model.MessageView;
 import org.apache.rocketmq.dashboard.model.request.MessageQuery;

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

@@ -17,7 +17,7 @@
 package org.apache.rocketmq.dashboard.controller;
 
 import javax.annotation.Resource;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
 import org.apache.rocketmq.dashboard.model.ConnectionInfo;
 import org.apache.rocketmq.dashboard.permisssion.Permission;
 import org.apache.rocketmq.dashboard.service.ProducerService;

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

@@ -0,0 +1,54 @@
+/*
+ * 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.controller;
+
+import org.apache.rocketmq.dashboard.permisssion.Permission;
+import org.apache.rocketmq.dashboard.service.ProxyService;
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+import javax.annotation.Resource;
+
+@Controller
+@RequestMapping("/proxy")
+@Permission
+public class ProxyController {
+    @Resource
+    private ProxyService proxyService;
+    @RequestMapping(value = "/homePage.query", method = RequestMethod.GET)
+    @ResponseBody
+    public Object homePage() {
+        return proxyService.getProxyHomePage();
+    }
+
+    @RequestMapping(value = "/addProxyAddr.do", method = RequestMethod.POST)
+    @ResponseBody
+    public Object addProxyAddr(@RequestParam String newProxyAddr) {
+        proxyService.addProxyAddrList(newProxyAddr);
+        return true;
+    }
+
+    @RequestMapping(value = "/updateProxyAddr.do", method = RequestMethod.POST)
+    @ResponseBody
+    public Object updateProxyAddr(@RequestParam String proxyAddr) {
+        proxyService.updateProxyAddrList(proxyAddr);
+        return true;
+    }
+}

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

@@ -56,6 +56,12 @@ public class TopicController {
         return topicService.fetchAllTopicList(skipSysProcess, skipRetryAndDlq);
     }
 
+    @RequestMapping(value = "/list.queryTopicType", method = RequestMethod.GET)
+    @ResponseBody
+    public Object listTopicType() {
+        return topicService.examineAllTopicType();
+    }
+
     @RequestMapping(value = "/stats.query", method = RequestMethod.GET)
     @ResponseBody
     public Object stats(@RequestParam String topic) {

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

@@ -20,7 +20,7 @@ import com.google.common.collect.Sets;
 import java.util.Collection;
 import java.util.HashSet;
 import org.apache.rocketmq.common.MQVersion;
-import org.apache.rocketmq.common.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
 
 public class ConnectionInfo extends Connection {
     private String versionDesc;

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

@@ -16,7 +16,7 @@
  */
 package org.apache.rocketmq.dashboard.model;
 
-import org.apache.rocketmq.common.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
 import com.google.common.collect.Lists;
 
 import java.util.List;

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

@@ -17,8 +17,8 @@
 package org.apache.rocketmq.dashboard.model;
 
 import lombok.Data;
-import org.apache.rocketmq.common.protocol.body.CMResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.CMResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 
 @Data
 public class DlqMessageResendResult {

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

@@ -16,8 +16,10 @@
  */
 package org.apache.rocketmq.dashboard.model;
 
-import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
-import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
+import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
+
+import java.util.List;
 
 public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
     private String group;
@@ -25,8 +27,11 @@ public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
     private int count;
     private ConsumeType consumeType;
     private MessageModel messageModel;
+    private List<String> address;
     private int consumeTps;
     private long diffTotal = -1;
+    private String subGroupType = "NORMAL";
+
 
     public String getGroup() {
         return group;
@@ -68,6 +73,22 @@ public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
         this.diffTotal = diffTotal;
     }
 
+    public List<String> getAddress() {
+        return address;
+    }
+
+    public void setAddress(List<String> address) {
+        this.address = address;
+    }
+
+    public String getSubGroupType() {
+        return subGroupType;
+    }
+
+    public void setSubGroupType(String subGroupType) {
+        this.subGroupType = subGroupType;
+    }
+
     @Override
     public int compareTo(GroupConsumeInfo o) {
         if (this.count != o.count) {

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

@@ -16,7 +16,7 @@
  */
 package org.apache.rocketmq.dashboard.model;
 
-import org.apache.rocketmq.common.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.springframework.beans.BeanUtils;
 

src/main/java/org/apache/rocketmq/dashboard/model/request/ConsumerConfigInfo.java

@@ -16,7 +16,7 @@
  */
 package org.apache.rocketmq.dashboard.model.request;
 
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 
 import java.util.List;
 

src/main/java/org/apache/rocketmq/dashboard/model/request/TopicConfigInfo.java

@@ -15,7 +15,6 @@
  * limitations under the License.
  */
 package org.apache.rocketmq.dashboard.model.request;
-
 import com.google.common.base.Objects;
 
 import java.util.List;
@@ -32,6 +31,7 @@ public class TopicConfigInfo {
     private int perm;
     private boolean order;
 
+    private String messageType;
     public List<String> getClusterNameList() {
         return clusterNameList;
     }
@@ -92,6 +92,18 @@ public class TopicConfigInfo {
         this.order = order;
     }
 
+
+    public String getMessageType() {
+        return messageType;
+    }
+
+    public void setMessageType(String messageType) {
+        this.messageType = messageType;
+    }
+
+
+
+
     @Override
     public boolean equals(Object o) {
         if (this == o)
@@ -103,12 +115,13 @@ public class TopicConfigInfo {
             readQueueNums == that.readQueueNums &&
             perm == that.perm &&
             order == that.order &&
-            Objects.equal(topicName, that.topicName);
+            Objects.equal(topicName, that.topicName) &&
+            Objects.equal(messageType, that.messageType);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hashCode(topicName, writeQueueNums, readQueueNums, perm, order);
+        return Objects.hashCode(topicName, writeQueueNums, readQueueNums, perm, order,messageType);
     }
 
 }

src/main/java/org/apache/rocketmq/dashboard/model/request/TopicTypeList.java

@@ -0,0 +1,45 @@
+/*
+ * 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.model.request;
+
+import java.util.List;
+
+public class TopicTypeList {
+    private List<String> topicNameList;
+    private List<String> messageTypeList;
+
+    public List<String> getTopicNameList() {
+        return topicNameList;
+    }
+
+    public void setTopicNameList(List<String> topicNameList) {
+        this.topicNameList = topicNameList;
+    }
+
+    public List<String> getMessageTypeList() {
+        return messageTypeList;
+    }
+
+    public void setMessageTypeList(List<String> messageTypeList) {
+        this.messageTypeList = messageTypeList;
+    }
+
+    public TopicTypeList(List<String> topicNameList, List<String> messageTypeList) {
+        this.topicNameList = topicNameList;
+        this.messageTypeList = messageTypeList;
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/model/request/TopicTypeMeta.java

@@ -0,0 +1,38 @@
+/*
+ * 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.model.request;
+
+public class TopicTypeMeta {
+    private String topicName;
+    private String messageType;
+
+    public String getTopicName() {
+        return topicName;
+    }
+
+    public void setTopicName(String topicName) {
+        this.topicName = topicName;
+    }
+
+    public String getMessageType() {
+        return messageType;
+    }
+
+    public void setMessageType(String messageType) {
+        this.messageType = messageType;
+    }
+}

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

@@ -19,8 +19,8 @@ package org.apache.rocketmq.dashboard.service;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import com.google.common.base.Throwables;
 import com.google.common.collect.Sets;
-import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import javax.annotation.Resource;
 import org.apache.commons.collections.CollectionUtils;
@@ -28,7 +28,7 @@ import org.apache.commons.collections.CollectionUtils;
 public abstract class AbstractCommonService {
     @Resource
     protected MQAdminExt mqAdminExt;
-    protected final Set<String> changeToBrokerNameSet(HashMap<String, Set<String>> clusterAddrTable,
+    protected final Set<String> changeToBrokerNameSet(Map<String, Set<String>> clusterAddrTable,
         List<String> clusterNameList, List<String> brokerNameList) {
         Set<String> finalBrokerNameList = Sets.newHashSet();
         if (CollectionUtils.isNotEmpty(clusterNameList)) {
@@ -38,7 +38,8 @@ public abstract class AbstractCommonService {
                 }
             }
             catch (Exception e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
             }
         }
         if (CollectionUtils.isNotEmpty(brokerNameList)) {

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

@@ -17,8 +17,8 @@
 
 package org.apache.rocketmq.dashboard.service;
 
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
+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;
@@ -31,12 +31,12 @@ import java.util.Map;
 import java.util.Set;
 
 public interface ConsumerService {
-    List<GroupConsumeInfo> queryGroupList(boolean skipSysGroup);
+    List<GroupConsumeInfo> queryGroupList(boolean skipSysGroup,String address);
 
-    GroupConsumeInfo queryGroup(String consumerGroup);
+    GroupConsumeInfo queryGroup(String consumerGroup, String address);
 
 
-    List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName);
+    List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName, String address);
 
     List<TopicConsumerInfo> queryConsumeStatsList(String topic, String groupName);
 
@@ -52,7 +52,7 @@ public interface ConsumerService {
 
     Set<String> fetchBrokerNameSetBySubscriptionGroup(String group);
 
-    ConsumerConnection getConsumerConnection(String consumerGroup);
+    ConsumerConnection getConsumerConnection(String consumerGroup, String address);
 
     ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack);
 }

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

@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.service;
 
 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 import org.apache.rocketmq.dashboard.model.MessagePage;
 import org.apache.rocketmq.dashboard.model.request.MessageQuery;
 import org.apache.rocketmq.tools.admin.api.MessageTrack;

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

@@ -17,7 +17,7 @@
 
 package org.apache.rocketmq.dashboard.service;
 
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
 
 public interface ProducerService {
     ProducerConnection getProducerConnection(String producerGroup, String topic);

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

@@ -0,0 +1,28 @@
+/*
+ * 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 java.util.Map;
+
+public interface ProxyService {
+
+    void addProxyAddrList(String proxyAddr);
+
+    void updateProxyAddrList(String proxyAddr);
+
+    Map<String, Object>  getProxyHomePage();
+}

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

@@ -19,10 +19,11 @@ package org.apache.rocketmq.dashboard.service;
 
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.dashboard.model.request.TopicTypeList;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
 
@@ -31,6 +32,8 @@ import java.util.List;
 public interface TopicService {
     TopicList fetchAllTopicList(boolean skipSysProcess, boolean skipRetryAndDlq);
 
+    TopicTypeList examineAllTopicType();
+
     TopicStatsTable stats(String topic);
 
     TopicRouteData route(String topic);

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

@@ -17,43 +17,16 @@
 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 org.apache.rocketmq.client.QueryResult;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.client.impl.MQAdminImpl;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.RequestCode;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.ClusterAclVersionInfo;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
-import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
-import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.common.message.MessageRequestMode;
 import org.apache.rocketmq.dashboard.util.JsonUtil;
 import org.apache.rocketmq.remoting.RemotingClient;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
@@ -62,13 +35,51 @@ import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
 import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
+import org.apache.rocketmq.remoting.protocol.body.QueryConsumeQueueResponseBody;
+import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
+import org.apache.rocketmq.remoting.protocol.subscription.GroupForbidden;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.apache.rocketmq.tools.admin.api.BrokerOperatorResult;
 import org.apache.rocketmq.tools.admin.api.MessageTrack;
-import org.joor.Reflect;
+import org.apache.rocketmq.tools.admin.common.AdminToolResult;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
+import java.io.UnsupportedEncodingException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
 import static org.apache.rocketmq.remoting.protocol.RemotingSerializable.decode;
 
 @Service
@@ -81,7 +92,7 @@ public class MQAdminExtImpl implements MQAdminExt {
     @Override
     public void updateBrokerConfig(String brokerAddr, Properties properties)
         throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
-        UnsupportedEncodingException, InterruptedException, MQBrokerException {
+        UnsupportedEncodingException, InterruptedException, MQBrokerException, MQClientException {
         MQAdminInstance.threadLocalMQAdminExt().updateBrokerConfig(brokerAddr, properties);
     }
 
@@ -128,7 +139,7 @@ public class MQAdminExtImpl implements MQAdminExt {
     }
 
     @Override
-    public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) {
+    public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) throws MQBrokerException {
         RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null);
         RemotingCommand response = null;
@@ -136,7 +147,8 @@ public class MQAdminExtImpl implements MQAdminExt {
             response = remotingClient.invokeSync(addr, request, 3000);
         }
         catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
         assert response != null;
         switch (response.getCode()) {
@@ -145,12 +157,12 @@ public class MQAdminExtImpl implements MQAdminExt {
                 return subscriptionGroupWrapper.getSubscriptionGroupTable().get(group);
             }
             default:
-                throw Throwables.propagate(new MQBrokerException(response.getCode(), response.getRemark()));
+                throw new MQBrokerException(response.getCode(), response.getRemark());
         }
     }
 
     @Override
-    public TopicConfig examineTopicConfig(String addr, String topic) {
+    public TopicConfig examineTopicConfig(String addr, String topic) throws MQBrokerException {
         RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
         RemotingCommand response = null;
@@ -158,7 +170,8 @@ public class MQAdminExtImpl implements MQAdminExt {
             response = remotingClient.invokeSync(addr, request, 3000);
         }
         catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
         switch (response.getCode()) {
             case ResponseCode.SUCCESS: {
@@ -166,7 +179,7 @@ public class MQAdminExtImpl implements MQAdminExt {
                 return topicConfigSerializeWrapper.getTopicConfigTable().get(topic);
             }
             default:
-                throw Throwables.propagate(new MQBrokerException(response.getCode(), response.getRemark()));
+                throw new MQBrokerException(response.getCode(), response.getRemark());
         }
     }
 
@@ -376,14 +389,14 @@ public class MQAdminExtImpl implements MQAdminExt {
     }
 
     @Override
-    public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum);
+    public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
+        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, attributes);
     }
 
     @Override
-    public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
+    public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes)
         throws MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag);
+        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag, attributes);
     }
 
     @Override
@@ -451,15 +464,18 @@ public class MQAdminExtImpl implements MQAdminExt {
         }
         catch (Exception e) {
         }
-        MQAdminImpl mqAdminImpl = MQAdminInstance.threadLocalMqClientInstance().getMQAdminImpl();
-        QueryResult qr = Reflect.on(mqAdminImpl).call("queryMessage", topic, msgId, 32,
-            MessageClientIDSetter.getNearlyTimeFromID(msgId).getTime() - 1000 * 60 * 60 * 13L, Long.MAX_VALUE, true).get();
-        if (qr != null && qr.getMessageList() != null && qr.getMessageList().size() > 0) {
-            return qr.getMessageList().get(0);
+
+        Set<String> clusterList = MQAdminInstance.threadLocalMQAdminExt().getTopicClusterList(topic);
+        if (clusterList == null || clusterList.isEmpty()) {
+            return MQAdminInstance.threadLocalMQAdminExt().queryMessage("", topic, msgId);
         }
-        else {
-            return null;
+        for (String name : clusterList) {
+            MessageExt messageExt = MQAdminInstance.threadLocalMQAdminExt().queryMessage(name, topic, msgId);
+            if (messageExt != null) {
+                return messageExt;
+            }
         }
+        return null;
     }
 
     @Override
@@ -572,4 +588,256 @@ public class MQAdminExtImpl implements MQAdminExt {
         String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
         return false;
     }
+
+    @Override
+    public void addBrokerToContainer(String brokerContainerAddr, String brokerConfig) throws InterruptedException,
+            MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'addBrokerToContainer'");
+    }
+
+    @Override
+    public void removeBrokerFromContainer(String brokerContainerAddr, String clusterName, String brokerName,
+            long brokerId) throws InterruptedException, MQBrokerException, RemotingTimeoutException,
+            RemotingSendRequestException, RemotingConnectException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'removeBrokerFromContainer'");
+    }
+
+    @Override
+    public void updateGlobalWhiteAddrConfig(String addr, String globalWhiteAddrs, String aclFileFullPath)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'updateGlobalWhiteAddrConfig'");
+    }
+
+    @Override
+    public TopicStatsTable examineTopicStats(String brokerAddr, String topic)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineTopicStats'");
+    }
+
+    @Override
+    public AdminToolResult<TopicStatsTable> examineTopicStatsConcurrent(String topic) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineTopicStatsConcurrent'");
+    }
+
+    @Override
+    public ConsumeStats examineConsumeStats(String brokerAddr, String consumerGroup, String topicName,
+            long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException, MQBrokerException {
+        // TODO Auto-generated method stub
+        return MQAdminInstance.threadLocalMQAdminExt().examineConsumeStats(brokerAddr, consumerGroup, topicName, timeoutMillis);
+    }
+
+    @Override
+    public AdminToolResult<ConsumeStats> examineConsumeStatsConcurrent(String consumerGroup, String topic) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStatsConcurrent'");
+    }
+
+    @Override
+    public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup, String brokerAddr)
+            throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        return MQAdminInstance.threadLocalMQAdminExt().examineConsumerConnectionInfo(consumerGroup, brokerAddr);
+    }
+
+    @Override
+    public ProducerTableInfo getAllProducerInfo(String brokerAddr)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getAllProducerInfo'");
+    }
+
+    @Override
+    public void deleteTopic(String topicName, String clusterName)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteTopic'");
+    }
+
+    @Override
+    public AdminToolResult<BrokerOperatorResult> deleteTopicInBrokerConcurrent(Set<String> addrs, String topic) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteTopicInBrokerConcurrent'");
+    }
+
+    @Override
+    public void deleteTopicInNameServer(Set<String> addrs, String clusterName, String topic)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteTopicInNameServer'");
+    }
+
+    @Override
+    public AdminToolResult<BrokerOperatorResult> resetOffsetNewConcurrent(String group, String topic, long timestamp) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'resetOffsetNewConcurrent'");
+    }
+
+    @Override
+    public TopicList queryTopicsByConsumer(String group)
+            throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryTopicsByConsumer'");
+    }
+
+    @Override
+    public AdminToolResult<TopicList> queryTopicsByConsumerConcurrent(String group) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryTopicsByConsumerConcurrent'");
+    }
+
+    @Override
+    public SubscriptionData querySubscription(String group, String topic)
+            throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'querySubscription'");
+    }
+
+    @Override
+    public AdminToolResult<List<QueueTimeSpan>> queryConsumeTimeSpanConcurrent(String topic, String group) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryConsumeTimeSpanConcurrent'");
+    }
+
+    @Override
+    public boolean deleteExpiredCommitLog(String cluster) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteExpiredCommitLog'");
+    }
+
+    @Override
+    public boolean deleteExpiredCommitLogByAddr(String addr) throws RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteExpiredCommitLogByAddr'");
+    }
+
+    @Override
+    public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack,
+            boolean metrics) throws RemotingException, MQClientException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getConsumerRunningInfo'");
+    }
+
+    @Override
+    public List<MessageTrack> messageTrackDetailConcurrent(MessageExt msg)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'messageTrackDetailConcurrent'");
+    }
+
+    @Override
+    public void setMessageRequestMode(String brokerAddr, String topic, String consumerGroup, MessageRequestMode mode,
+            int popWorkGroupSize, long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
+            RemotingSendRequestException, RemotingConnectException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'setMessageRequestMode'");
+    }
+
+    @Override
+    public long searchOffset(String brokerAddr, String topicName, int queueId, long timestamp, long timeoutMillis)
+            throws RemotingException, MQBrokerException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'searchOffset'");
+    }
+
+    @Override
+    public void resetOffsetByQueueId(String brokerAddr, String consumerGroup, String topicName, int queueId,
+            long resetOffset) throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'resetOffsetByQueueId'");
+    }
+
+    @Override
+    public void createStaticTopic(String addr, String defaultTopic, TopicConfig topicConfig,
+            TopicQueueMappingDetail mappingDetail, boolean force)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'createStaticTopic'");
+    }
+
+    @Override
+    public GroupForbidden updateAndGetGroupReadForbidden(String brokerAddr, String groupName, String topicName,
+            Boolean readable) throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'updateAndGetGroupReadForbidden'");
+    }
+
+    @Override
+    public MessageExt queryMessage(String clusterName, String topic, String msgId)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryMessage'");
+    }
+
+    @Override
+    public HARuntimeInfo getBrokerHAStatus(String brokerAddr) throws RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getBrokerHAStatus'");
+    }
+
+    @Override
+    public BrokerReplicasInfo getInSyncStateData(String controllerAddress, List<String> brokers)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getInSyncStateData'");
+    }
+
+    @Override
+    public EpochEntryCache getBrokerEpochCache(String brokerAddr)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getBrokerEpochCache'");
+    }
+
+    @Override
+    public GetMetaDataResponseHeader getControllerMetaData(String controllerAddr)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getControllerMetaData'");
+    }
+
+    @Override
+    public void resetMasterFlushOffset(String brokerAddr, long masterFlushOffset) throws InterruptedException,
+            MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'resetMasterFlushOffset'");
+    }
+
+    @Override
+    public Map<String, Properties> getControllerConfig(List<String> controllerServers)
+            throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException, MQClientException, UnsupportedEncodingException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getControllerConfig'");
+    }
+
+    @Override
+    public void updateControllerConfig(Properties properties, List<String> controllers)
+            throws InterruptedException, RemotingConnectException, UnsupportedEncodingException,
+            RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'updateControllerConfig'");
+    }
+
+    @Override
+    public ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, String brokerName,
+            String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'electMaster'");
+    }
+
+    @Override
+    public void cleanControllerBrokerData(String controllerAddr, String clusterName, String brokerName,
+            String brokerAddr, boolean isCleanLivingBroker)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'cleanControllerBrokerData'");
+    }
 }

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

@@ -0,0 +1,28 @@
+/*
+ * 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.client;
+
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+
+public interface ProxyAdmin {
+
+    ConsumerConnection examineConsumerConnectionInfo(String addr, String consumerGroup) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException;
+}

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

@@ -0,0 +1,60 @@
+/*
+ * 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.client;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerConnectionListRequestHeader;
+import org.apache.rocketmq.tools.admin.MQAdminExt;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import static org.apache.rocketmq.remoting.protocol.RequestCode.GET_CONSUMER_CONNECTION_LIST;
+
+@Slf4j
+@Service
+public class ProxyAdminImpl implements ProxyAdmin {
+    @Autowired
+    private GenericObjectPool<MQAdminExt> mqAdminExtPool;
+
+    @Override
+    public ConsumerConnection examineConsumerConnectionInfo(String addr, String consumerGroup) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
+        try {
+            MQAdminInstance.createMQAdmin(mqAdminExtPool);
+            RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
+            GetConsumerConnectionListRequestHeader requestHeader = new GetConsumerConnectionListRequestHeader();
+            requestHeader.setConsumerGroup(consumerGroup);
+            RemotingCommand request = RemotingCommand.createRequestCommand(GET_CONSUMER_CONNECTION_LIST, requestHeader);
+            RemotingCommand response = remotingClient.invokeSync(addr, request, 3000);
+            switch (response.getCode()) {
+                case 0:
+                    return ConsumerConnection.decode(response.getBody(), ConsumerConnection.class);
+                default:
+                    throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
+            }
+        } finally {
+            MQAdminInstance.returnMQAdmin(mqAdminExtPool);
+        }
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/service/impl/AclServiceImpl.java

@@ -36,8 +36,8 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.dashboard.model.request.AclRequest;
 import org.apache.rocketmq.dashboard.service.AbstractCommonService;
 import org.apache.rocketmq.dashboard.service.AclService;
@@ -68,7 +68,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
             }
         } catch (Exception e) {
             log.error("getAclConfig error.", e);
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         AclConfig aclConfig = new AclConfig();
         aclConfig.setGlobalWhiteAddrs(Collections.emptyList());
@@ -100,7 +101,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
 
     }
@@ -116,7 +118,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 log.info("Delete acl [{}] from broker [{}] complete", config.getAccessKey(), addr);
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -142,7 +145,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -174,7 +178,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 }
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -206,7 +211,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 }
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -249,7 +255,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 }
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
 
     }
@@ -261,7 +268,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -281,7 +289,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(aclConfig.getGlobalWhiteAddrs(), ","));
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -297,7 +306,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(aclConfig.getGlobalWhiteAddrs(), ","));
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -311,7 +321,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                 mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(whiteList, ","));
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 

src/main/java/org/apache/rocketmq/dashboard/service/impl/ClusterServiceImpl.java

@@ -17,9 +17,10 @@
 
 package org.apache.rocketmq.dashboard.service.impl;
 
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.apache.rocketmq.dashboard.service.ClusterService;
 import org.apache.rocketmq.dashboard.util.JsonUtil;
@@ -30,8 +31,10 @@ import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;
 
 @Service
 public class ClusterServiceImpl implements ClusterService {
@@ -56,10 +59,14 @@ public class ClusterServiceImpl implements ClusterService {
             }
             resultMap.put("clusterInfo", clusterInfo);
             resultMap.put("brokerServer", brokerServer);
+            // add messageType
+            resultMap.put("messageTypes", Arrays.stream(TopicMessageType.values()).sorted()
+                    .collect(Collectors.toMap(TopicMessageType::getValue, messageType ->String.format("MESSAGE_TYPE_%s",messageType.getValue()))));
             return resultMap;
         }
         catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
     }
 
@@ -70,7 +77,8 @@ public class ClusterServiceImpl implements ClusterService {
             return mqAdminExt.getBrokerConfig(brokerAddr);
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 }

src/main/java/org/apache/rocketmq/dashboard/service/impl/ConsumerServiceImpl.java

@@ -23,12 +23,22 @@ import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionHandler;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import javax.annotation.Resource;
 import org.apache.commons.collections.CollectionUtils;
@@ -36,18 +46,20 @@ import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
+import org.apache.rocketmq.dashboard.service.client.ProxyAdmin;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.common.utils.ThreadUtils;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.model.ConsumerGroupRollBackStat;
 import org.apache.rocketmq.dashboard.model.GroupConsumeInfo;
@@ -60,19 +72,46 @@ import org.apache.rocketmq.dashboard.service.AbstractCommonService;
 import org.apache.rocketmq.dashboard.service.ConsumerService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.stereotype.Service;
 
-import static com.google.common.base.Throwables.propagate;
-
 @Service
-public class ConsumerServiceImpl extends AbstractCommonService implements ConsumerService {
+public class ConsumerServiceImpl extends AbstractCommonService implements ConsumerService, InitializingBean, DisposableBean {
     private Logger logger = LoggerFactory.getLogger(ConsumerServiceImpl.class);
 
+    @Resource
+    protected ProxyAdmin proxyAdmin;
     @Resource
     private RMQConfigure configure;
 
     private static final Set<String> SYSTEM_GROUP_SET = new HashSet<>();
 
+    private ExecutorService executorService;
+
+    @Override
+    public void afterPropertiesSet() {
+        Runtime runtime = Runtime.getRuntime();
+        int corePoolSize = Math.max(10, runtime.availableProcessors() * 2);
+        int maximumPoolSize = Math.max(20, runtime.availableProcessors() * 2);
+        ThreadFactory threadFactory = new ThreadFactory() {
+            private final AtomicLong threadIndex = new AtomicLong(0);
+
+            @Override
+            public Thread newThread(Runnable r) {
+                return new Thread(r, "QueryGroup_" + this.threadIndex.incrementAndGet());
+            }
+        };
+        RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardOldestPolicy();
+        this.executorService = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 60L, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(5000), threadFactory, handler);
+    }
+
+    @Override
+    public void destroy() {
+        ThreadUtils.shutdownGracefully(executorService, 10L, TimeUnit.SECONDS);
+    }
+
     static {
         SYSTEM_GROUP_SET.add(MixAll.TOOLS_CONSUMER_GROUP);
         SYSTEM_GROUP_SET.add(MixAll.FILTERSRV_CONSUMER_GROUP);
@@ -85,22 +124,47 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
     }
 
     @Override
-    public List<GroupConsumeInfo> queryGroupList(boolean skipSysGroup) {
-        Set<String> consumerGroupSet = Sets.newHashSet();
+    public List<GroupConsumeInfo> queryGroupList(boolean skipSysGroup, String address) {
+        HashMap<String, List<String>> consumerGroupMap = Maps.newHashMap();
         try {
             ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
             for (BrokerData brokerData : clusterInfo.getBrokerAddrTable().values()) {
                 SubscriptionGroupWrapper subscriptionGroupWrapper = mqAdminExt.getAllSubscriptionGroup(brokerData.selectBrokerAddr(), 3000L);
-                consumerGroupSet.addAll(subscriptionGroupWrapper.getSubscriptionGroupTable().keySet());
+                for (String groupName : subscriptionGroupWrapper.getSubscriptionGroupTable().keySet()) {
+                    if (!consumerGroupMap.containsKey(groupName)) {
+                        consumerGroupMap.putIfAbsent(groupName, new ArrayList<>());
+                    }
+                    List<String> addresses = consumerGroupMap.get(groupName);
+                    addresses.add(brokerData.selectBrokerAddr());
+                    consumerGroupMap.put(groupName, addresses);
+                }
             }
+        } catch (Exception err) {
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
-        catch (Exception err) {
-            throw Throwables.propagate(err);
+        List<GroupConsumeInfo> groupConsumeInfoList = Collections.synchronizedList(Lists.newArrayList());
+        CountDownLatch countDownLatch = new CountDownLatch(consumerGroupMap.size());
+        for (Map.Entry<String, List<String>> entry : consumerGroupMap.entrySet()) {
+            String consumerGroup = entry.getKey();
+            executorService.submit(() -> {
+                try {
+                    GroupConsumeInfo consumeInfo = queryGroup(consumerGroup, address);
+                    consumeInfo.setAddress(entry.getValue());
+                    groupConsumeInfoList.add(consumeInfo);
+                } catch (Exception e) {
+                    logger.error("queryGroup exception, consumerGroup: {}", consumerGroup, e);
+                } finally {
+                    countDownLatch.countDown();
+                }
+            });
         }
-        List<GroupConsumeInfo> groupConsumeInfoList = Lists.newArrayList();
-        for (String consumerGroup : consumerGroupSet) {
-            groupConsumeInfoList.add(queryGroup(consumerGroup));
+        try {
+            countDownLatch.await(30, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            logger.error("query consumerGroup countDownLatch await Exception", e);
         }
+
         if (!skipSysGroup) {
             groupConsumeInfoList.stream().map(group -> {
                 if (SYSTEM_GROUP_SET.contains(group.getGroup())) {
@@ -114,7 +178,7 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
     }
 
     @Override
-    public GroupConsumeInfo queryGroup(String consumerGroup) {
+    public GroupConsumeInfo queryGroup(String consumerGroup, String address) {
         GroupConsumeInfo groupConsumeInfo = new GroupConsumeInfo();
         try {
             ConsumeStats consumeStats = null;
@@ -126,14 +190,28 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             }
 
             ConsumerConnection consumerConnection = null;
+            boolean isFifoType = examineSubscriptionGroupConfig(consumerGroup)
+                    .stream().map(ConsumerConfigInfo::getSubscriptionGroupConfig)
+                    .allMatch(SubscriptionGroupConfig::isConsumeMessageOrderly);
+
             try {
-                consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
-            }
-            catch (Exception e) {
+                if (StringUtils.isNotEmpty(address)) {
+                    consumerConnection = proxyAdmin.examineConsumerConnectionInfo(address, consumerGroup);
+                } else {
+                    consumerConnection = mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
+                }
+            } catch (Exception e) {
                 logger.warn("examineConsumeStats exception to consumerGroup {}, response [{}]", consumerGroup, e.getMessage());
             }
 
             groupConsumeInfo.setGroup(consumerGroup);
+            if (SYSTEM_GROUP_SET.contains(consumerGroup)) {
+                groupConsumeInfo.setSubGroupType("SYSTEM");
+            } else if (isFifoType) {
+                groupConsumeInfo.setSubGroupType("FIFO");
+            } else {
+                groupConsumeInfo.setSubGroupType("NORMAL");
+            }
 
             if (consumeStats != null) {
                 groupConsumeInfo.setConsumeTps((int)consumeStats.getConsumeTps());
@@ -155,8 +233,18 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
     }
 
     @Override
-    public List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName) {
-        return queryConsumeStatsList(null, groupName);
+    public List<TopicConsumerInfo> queryConsumeStatsListByGroupName(String groupName, String address) {
+        ConsumeStats consumeStats;
+        String topic = null;
+        try {
+            String[] addresses = address.split(",");
+            String addr = addresses[0];
+            consumeStats = mqAdminExt.examineConsumeStats(addr, groupName, null, 3000);
+        } catch (Exception e) {
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
+        }
+        return toTopicConsumerInfoList(topic, consumeStats, groupName);
     }
 
     @Override
@@ -166,8 +254,13 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             consumeStats = mqAdminExt.examineConsumeStats(groupName, topic);
         }
         catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
+        return toTopicConsumerInfoList(topic, consumeStats, groupName);
+    }
+
+    private List<TopicConsumerInfo> toTopicConsumerInfoList(String topic, ConsumeStats consumeStats, String groupName) {
         List<MessageQueue> mqList = Lists.newArrayList(Iterables.filter(consumeStats.getOffsetTable().keySet(), new Predicate<MessageQueue>() {
             @Override
             public boolean apply(MessageQueue o) {
@@ -226,7 +319,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             return group2ConsumerInfoMap;
         }
         catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -289,7 +383,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             }
         }
         catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         return consumerConfigInfoList;
     }
@@ -314,7 +409,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             }
         }
         catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         return true;
     }
@@ -341,7 +437,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             }
         }
         catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
         return true;
     }
@@ -356,19 +453,22 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             }
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         return brokerNameSet;
 
     }
 
     @Override
-    public ConsumerConnection getConsumerConnection(String consumerGroup) {
+    public ConsumerConnection getConsumerConnection(String consumerGroup, String address) {
         try {
-            return mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
-        }
-        catch (Exception e) {
-            throw Throwables.propagate(e);
+            String[] addresses = address.split(",");
+            String addr = addresses[0];
+            return mqAdminExt.examineConsumerConnectionInfo(consumerGroup, addr);
+        } catch (Exception e) {
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -378,7 +478,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             return mqAdminExt.getConsumerRunningInfo(consumerGroup, clientId, jstack);
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 }

src/main/java/org/apache/rocketmq/dashboard/service/impl/DashboardCollectServiceImpl.java

@@ -107,7 +107,8 @@ public class DashboardCollectServiceImpl implements DashboardCollectService {
             strings = Files.readLines(file, Charsets.UTF_8);
         }
         catch (IOException e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         StringBuffer sb = new StringBuffer();
         for (String string : strings) {

src/main/java/org/apache/rocketmq/dashboard/service/impl/DlqMessageServiceImpl.java

@@ -25,8 +25,8 @@ import javax.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 import org.apache.rocketmq.dashboard.model.DlqMessageResendResult;
 import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
 import org.apache.rocketmq.dashboard.model.MessagePage;
@@ -62,10 +62,12 @@ public class DlqMessageServiceImpl implements DlqMessageService {
                 && e.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
                 return new MessagePage(new PageImpl<>(messageViews, page, 0), query.getTaskId());
             } else {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         return messageService.queryMessageByPage(query);
     }

src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java

@@ -37,9 +37,9 @@ import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.exception.ServiceException;
 import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
@@ -115,7 +115,8 @@ public class MessageServiceImpl implements MessageService {
             if (err instanceof MQClientException) {
                 throw new ServiceException(-1, ((MQClientException) err).getErrorMessage());
             }
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
     }
 
@@ -185,7 +186,8 @@ public class MessageServiceImpl implements MessageService {
             });
             return messageViewList;
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         } finally {
             consumer.shutdown();
         }
@@ -209,7 +211,8 @@ public class MessageServiceImpl implements MessageService {
             try {
                 return mqAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
             } catch (Exception e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
             }
         }
 
@@ -223,7 +226,8 @@ public class MessageServiceImpl implements MessageService {
                 return mqAdminExt.consumeMessageDirectly(consumerGroup, connection.getClientId(), topic, msgId);
             }
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         throw new IllegalStateException("NO CONSUMER");
 
@@ -272,7 +276,7 @@ public class MessageServiceImpl implements MessageService {
             int idx = 0;
             for (MessageQueue messageQueue : messageQueues) {
                 Long minOffset = consumer.searchOffset(messageQueue, query.getBegin());
-                Long maxOffset = consumer.searchOffset(messageQueue, query.getEnd()) + 1;
+                Long maxOffset = consumer.searchOffset(messageQueue, query.getEnd());
                 queueOffsetInfos.add(new QueueOffsetInfo(idx++, minOffset, maxOffset, minOffset, minOffset, messageQueue));
             }
 
@@ -388,7 +392,8 @@ public class MessageServiceImpl implements MessageService {
             PageImpl<MessageView> page = new PageImpl<>(messageViews, query.page(), total);
             return new MessagePageTask(page, queueOffsetInfos);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         } finally {
             consumer.shutdown();
         }
@@ -455,7 +460,8 @@ public class MessageServiceImpl implements MessageService {
             }
             return new PageImpl<>(messageViews, query.page(), total);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         } finally {
             consumer.shutdown();
         }

src/main/java/org/apache/rocketmq/dashboard/service/impl/MonitorServiceImpl.java

@@ -82,7 +82,8 @@ public class MonitorServiceImpl implements MonitorService {
             MixAll.string2File(dataStr, path);
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 

src/main/java/org/apache/rocketmq/dashboard/service/impl/ProducerServiceImpl.java

@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.service.impl;
 
 import com.google.common.base.Throwables;
 import javax.annotation.Resource;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
 import org.apache.rocketmq.dashboard.service.ProducerService;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.springframework.stereotype.Service;
@@ -35,7 +35,8 @@ public class ProducerServiceImpl implements ProducerService {
             return mqAdminExt.examineProducerConnectionInfo(producerGroup, topic);
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 }

src/main/java/org/apache/rocketmq/dashboard/service/impl/ProxyServiceImpl.java

@@ -0,0 +1,59 @@
+/*
+ * 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.impl;
+
+import com.google.common.collect.Maps;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.dashboard.config.RMQConfigure;
+import org.apache.rocketmq.dashboard.service.ProxyService;
+import org.apache.rocketmq.dashboard.service.client.ProxyAdmin;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+@Service
+public class ProxyServiceImpl implements ProxyService {
+    @Resource
+    protected ProxyAdmin proxyAdmin;
+    @Resource
+    private RMQConfigure configure;
+
+    @Override
+    public void addProxyAddrList(String proxyAddr) {
+        List<String> proxyAddrs = configure.getProxyAddrs();
+        if (proxyAddrs != null && !proxyAddrs.contains(proxyAddr)) {
+            proxyAddrs.add(proxyAddr);
+        }
+        configure.setProxyAddrs(proxyAddrs);
+    }
+
+    @Override
+    public void updateProxyAddrList(String proxyAddr) {
+        configure.setProxyAddr(proxyAddr);
+    }
+
+    @Override
+    public Map<String, Object> getProxyHomePage() {
+        Map<String, Object> homePageInfoMap = Maps.newHashMap();
+        homePageInfoMap.put("currentProxyAddr", configure.getProxyAddr());
+        homePageInfoMap.put("proxyAddrList", configure.getProxyAddrs());
+        return homePageInfoMap;
+    }
+}

src/main/java/org/apache/rocketmq/dashboard/service/impl/TopicServiceImpl.java

@@ -18,43 +18,57 @@
 package org.apache.rocketmq.dashboard.service.impl;
 
 import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
-import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.TransactionListener;
+import org.apache.rocketmq.client.producer.TransactionMQProducer;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
 import org.apache.rocketmq.client.trace.TraceContext;
 import org.apache.rocketmq.client.trace.TraceDispatcher;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
 import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
+import org.apache.rocketmq.dashboard.model.request.TopicTypeList;
+import org.apache.rocketmq.dashboard.model.request.TopicTypeMeta;
 import org.apache.rocketmq.dashboard.service.AbstractCommonService;
 import org.apache.rocketmq.dashboard.service.TopicService;
 import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.tools.command.CommandUtil;
 import org.joor.Reflect;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;
 
 @Service
 public class TopicServiceImpl extends AbstractCommonService implements TopicService {
@@ -68,23 +82,59 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
             TopicList allTopics = mqAdminExt.fetchAllTopicList();
             TopicList sysTopics = getSystemTopicList();
             Set<String> topics =
-                allTopics.getTopicList().stream().map(topic -> {
-                    if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
-                        topic = String.format("%s%s", "%SYS%", topic);
-                    }
-                    return topic;
-                }).filter(topic -> {
-                    if (skipRetryAndDlq) {
-                        return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
-                            || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
-                    }
-                    return true;
-                }).collect(Collectors.toSet());
+                    allTopics.getTopicList().stream().map(topic -> {
+                        if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
+                            topic = String.format("%s%s", "%SYS%", topic);
+                        }
+                        return topic;
+                    }).filter(topic -> {
+                        if (skipRetryAndDlq) {
+                            return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
+                                    || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
+                        }
+                        return true;
+                    }).collect(Collectors.toSet());
             allTopics.getTopicList().clear();
             allTopics.getTopicList().addAll(topics);
             return allTopics;
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public TopicTypeList examineAllTopicType() {
+        ArrayList<TopicTypeMeta> topicTypes = new ArrayList<>();
+        ArrayList<String> names = new ArrayList<>();
+        ArrayList<String> messageTypes = new ArrayList<>();
+        TopicList topicList = fetchAllTopicList(false, false);
+        checkTopicType(topicList, topicTypes);
+        topicTypes.sort((t1, t2) -> t1.getTopicName().compareTo(t2.getTopicName()));
+        for (TopicTypeMeta topicTypeMeta : topicTypes) {
+            names.add(topicTypeMeta.getTopicName());
+            messageTypes.add(topicTypeMeta.getMessageType());
+        }
+        return new TopicTypeList(names, messageTypes);
+    }
+
+    private void checkTopicType(TopicList topicList, ArrayList<TopicTypeMeta> topicTypes) {
+        for (String topicName : topicList.getTopicList()) {
+            TopicTypeMeta topicType = new TopicTypeMeta();
+            topicType.setTopicName(topicName);
+            if (topicName.startsWith("%R")) {
+                topicType.setMessageType("RETRY");
+            } else if (topicName.startsWith("%D")) {
+                topicType.setMessageType("DELAY");
+            } else if (topicName.startsWith("%S")) {
+                topicType.setMessageType("SYSTEM");
+            } else {
+                List<TopicConfigInfo> topicConfigInfos = examineTopicConfig(topicName);
+                if (!CollectionUtils.isEmpty(topicConfigInfos)) {
+                    topicType.setMessageType(topicConfigInfos.get(0).getMessageType());
+                }
+            }
+            topicTypes.add(topicType);
         }
     }
 
@@ -93,7 +143,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         try {
             return mqAdminExt.examineTopicStats(topic);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -102,7 +153,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         try {
             return mqAdminExt.examineTopicRouteInfo(topic);
         } catch (Exception ex) {
-            throw Throwables.propagate(ex);
+            Throwables.throwIfUnchecked(ex);
+            throw new RuntimeException(ex);
         }
     }
 
@@ -111,7 +163,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         try {
             return mqAdminExt.queryTopicConsumeByWho(topic);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -119,14 +172,20 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
     public void createOrUpdate(TopicConfigInfo topicCreateOrUpdateRequest) {
         TopicConfig topicConfig = new TopicConfig();
         BeanUtils.copyProperties(topicCreateOrUpdateRequest, topicConfig);
+        String messageType = topicCreateOrUpdateRequest.getMessageType();
+        if (StringUtils.isBlank(messageType)) {
+            messageType = TopicMessageType.NORMAL.name();
+        }
+        topicConfig.setAttributes(ImmutableMap.of("+".concat(TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName()), messageType));
         try {
             ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
             for (String brokerName : changeToBrokerNameSet(clusterInfo.getClusterAddrTable(),
-                topicCreateOrUpdateRequest.getClusterNameList(), topicCreateOrUpdateRequest.getBrokerNameList())) {
+                    topicCreateOrUpdateRequest.getClusterNameList(), topicCreateOrUpdateRequest.getBrokerNameList())) {
                 mqAdminExt.createAndUpdateTopicConfig(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(), topicConfig);
             }
         } catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
     }
 
@@ -137,7 +196,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
             clusterInfo = mqAdminExt.examineBrokerClusterInfo();
             return mqAdminExt.examineTopicConfig(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(), topic);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -150,6 +210,11 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
             TopicConfig topicConfig = examineTopicConfig(topic, brokerData.getBrokerName());
             BeanUtils.copyProperties(topicConfig, topicConfigInfo);
             topicConfigInfo.setBrokerNameList(Lists.newArrayList(brokerData.getBrokerName()));
+            String messageType = topicConfig.getAttributes().get(TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName());
+            if (StringUtils.isBlank(messageType)) {
+                messageType = TopicMessageType.UNSPECIFIED.name();
+            }
+            topicConfigInfo.setMessageType(messageType);
             topicConfigInfoList.add(topicConfigInfo);
         }
         return topicConfigInfoList;
@@ -170,7 +235,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
             }
             mqAdminExt.deleteTopicInNameServer(nameServerSet, topic);
         } catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
         return true;
     }
@@ -181,7 +247,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         try {
             clusterInfo = mqAdminExt.examineBrokerClusterInfo();
         } catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
         for (String clusterName : clusterInfo.getClusterAddrTable().keySet()) {
             deleteTopic(topic, clusterName);
@@ -197,11 +264,13 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
             try {
                 clusterInfo = mqAdminExt.examineBrokerClusterInfo();
             } catch (Exception e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
             }
             mqAdminExt.deleteTopicInBroker(Sets.newHashSet(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr()), topic);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
         return true;
     }
@@ -216,6 +285,12 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         return defaultMQProducer;
     }
 
+    public TransactionMQProducer buildTransactionMQProducer(String producerGroup, RPCHook rpcHook, boolean traceEnabled) {
+        TransactionMQProducer defaultMQProducer = new TransactionMQProducer(null, producerGroup, rpcHook, traceEnabled, TopicValidator.RMQ_SYS_TRACE_TOPIC);
+        defaultMQProducer.setUseTLS(configure.isUseTLS());
+        return defaultMQProducer;
+    }
+
     private TopicList getSystemTopicList() {
         RPCHook rpcHook = null;
         boolean isEnableAcl = !StringUtils.isEmpty(configure.getAccessKey()) && !StringUtils.isEmpty(configure.getSecretKey());
@@ -230,7 +305,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
             producer.start();
             return producer.getDefaultMQProducerImpl().getmQClientFactory().getMQClientAPIImpl().getSystemTopicList(20000L);
         } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         } finally {
             producer.shutdown();
         }
@@ -238,31 +314,61 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
 
     @Override
     public SendResult sendTopicMessageRequest(SendTopicMessageRequest sendTopicMessageRequest) {
-        DefaultMQProducer producer = null;
+        List<TopicConfigInfo> topicConfigInfos = examineTopicConfig(sendTopicMessageRequest.getTopic());
+        String messageType = topicConfigInfos.get(0).getMessageType();
         AclClientRPCHook rpcHook = null;
         if (configure.isACLEnabled()) {
             rpcHook = new AclClientRPCHook(new SessionCredentials(
-                configure.getAccessKey(),
-                configure.getSecretKey()
+                    configure.getAccessKey(),
+                    configure.getSecretKey()
             ));
         }
-        producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
-        producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
-        producer.setNamesrvAddr(configure.getNamesrvAddr());
-        try {
-            producer.start();
-            Message msg = new Message(sendTopicMessageRequest.getTopic(),
-                sendTopicMessageRequest.getTag(),
-                sendTopicMessageRequest.getKey(),
-                sendTopicMessageRequest.getMessageBody().getBytes()
-            );
-            return producer.send(msg);
-        } catch (Exception e) {
-            throw Throwables.propagate(e);
-        } finally {
-            waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
-            producer.shutdown();
+        if (TopicMessageType.TRANSACTION.getValue().equals(messageType)) {
+            // transaction message
+            TransactionListener transactionListener = new TransactionListenerImpl();
+
+            TransactionMQProducer producer = buildTransactionMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
+            producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
+            producer.setNamesrvAddr(configure.getNamesrvAddr());
+            producer.setTransactionListener(transactionListener);
+            try {
+                producer.start();
+                Message msg = new Message(sendTopicMessageRequest.getTopic(),
+                        sendTopicMessageRequest.getTag(),
+                        sendTopicMessageRequest.getKey(),
+                        sendTopicMessageRequest.getMessageBody().getBytes()
+                );
+                return producer.sendMessageInTransaction(msg, null);
+            } catch (Exception e) {
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
+            } finally {
+                waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
+                producer.shutdown();
+            }
+        } else {
+            // no transaction message
+            DefaultMQProducer producer = null;
+            producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
+            producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
+            producer.setNamesrvAddr(configure.getNamesrvAddr());
+            try {
+                producer.start();
+                Message msg = new Message(sendTopicMessageRequest.getTopic(),
+                        sendTopicMessageRequest.getTag(),
+                        sendTopicMessageRequest.getKey(),
+                        sendTopicMessageRequest.getMessageBody().getBytes()
+                );
+                return producer.send(msg);
+            } catch (Exception e) {
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
+            } finally {
+                waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
+                producer.shutdown();
+            }
         }
+
     }
 
     private void waitSendTraceFinish(DefaultMQProducer producer, boolean traceEnabled) {
@@ -284,4 +390,20 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         } catch (Exception ignore) {
         }
     }
+
+    static class TransactionListenerImpl implements TransactionListener {
+        private AtomicInteger transactionIndex = new AtomicInteger(0);
+
+        private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
+
+        @Override
+        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
+            return LocalTransactionState.COMMIT_MESSAGE;
+        }
+
+        @Override
+        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
+            return LocalTransactionState.COMMIT_MESSAGE;
+        }
+    }
 }

src/main/java/org/apache/rocketmq/dashboard/task/CollectTaskRunnble.java

@@ -24,12 +24,12 @@ import java.util.List;
 import java.util.concurrent.ExecutionException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.common.stats.Stats;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.service.DashboardCollectService;
-import org.apache.rocketmq.store.stats.BrokerStatsManager;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.apache.rocketmq.tools.command.stats.StatsAllSubCommand;
 
@@ -63,7 +63,7 @@ public class CollectTaskRunnble implements Runnable {
                 String masterAddr = bd.getBrokerAddrs().get(MixAll.MASTER_ID);
                 if (masterAddr != null) {
                     try {
-                        BrokerStatsData bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.TOPIC_PUT_NUMS, topic);
+                        BrokerStatsData bsd = mqAdminExt.viewBrokerStatsData(masterAddr, Stats.TOPIC_PUT_NUMS, topic);
                         inTPS += bsd.getStatsMinute().getTps();
                         inMsgCntToday += StatsAllSubCommand.compute24HourSum(bsd);
                     } catch (Exception e) {
@@ -78,7 +78,7 @@ public class CollectTaskRunnble implements Runnable {
                         if (masterAddr != null) {
                             try {
                                 String statsKey = String.format("%s@%s", topic, group);
-                                BrokerStatsData bsd = mqAdminExt.viewBrokerStatsData(masterAddr, BrokerStatsManager.GROUP_GET_NUMS, statsKey);
+                                BrokerStatsData bsd = mqAdminExt.viewBrokerStatsData(masterAddr, Stats.GROUP_GET_NUMS, statsKey);
                                 outTPS += bsd.getStatsMinute().getTps();
                                 outMsgCntToday += StatsAllSubCommand.compute24HourSum(bsd);
                             } catch (Exception e) {
@@ -93,7 +93,8 @@ public class CollectTaskRunnble implements Runnable {
             try {
                 list = dashboardCollectService.getTopicMap().get(topic);
             } catch (ExecutionException e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
             }
             if (null == list) {
                 list = Lists.newArrayList();

src/main/java/org/apache/rocketmq/dashboard/task/DashboardCollectTask.java

@@ -34,10 +34,10 @@ import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import javax.annotation.Resource;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.service.DashboardCollectService;
@@ -84,7 +84,8 @@ public class DashboardCollectTask {
             }
         }
         catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
         }
     }
 
@@ -128,7 +129,8 @@ public class DashboardCollectTask {
             log.debug("Broker Collected Data in memory = {}" + JsonUtil.obj2String(dashboardCollectService.getBrokerMap().asMap()));
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -144,10 +146,12 @@ public class DashboardCollectTask {
                 Thread.sleep(1000);
             }
             catch (InterruptedException e1) {
-                throw Throwables.propagate(e1);
+                Throwables.throwIfUnchecked(e1);
+                throw new RuntimeException(e1);
             }
             fetchBrokerRuntimeStats(brokerAddr, retryTime - 1);
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -197,7 +201,8 @@ public class DashboardCollectTask {
 
         }
         catch (IOException e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -250,7 +255,7 @@ public class DashboardCollectTask {
 
     private void addSystemTopic() throws Exception {
         ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
-        HashMap<String, Set<String>> clusterTable = clusterInfo.getClusterAddrTable();
+        Map<String, Set<String>> clusterTable = clusterInfo.getClusterAddrTable();
         for (Map.Entry<String, Set<String>> entry : clusterTable.entrySet()) {
             String clusterName = entry.getKey();
             TopicValidator.addSystemTopic(clusterName);

src/main/java/org/apache/rocketmq/dashboard/task/MonitorTask.java

@@ -40,7 +40,7 @@ public class MonitorTask {
 //    @Scheduled(cron = "* * * * * ?")
     public void scanProblemConsumeGroup() {
         for (Map.Entry<String, ConsumerMonitorConfig> configEntry : monitorService.queryConsumerMonitorConfig().entrySet()) {
-            GroupConsumeInfo consumeInfo = consumerService.queryGroup(configEntry.getKey());
+            GroupConsumeInfo consumeInfo = consumerService.queryGroup(configEntry.getKey(), null);
             if (consumeInfo.getCount() < configEntry.getValue().getMinCount() || consumeInfo.getDiffTotal() > configEntry.getValue().getMaxDiffTotal()) {
                 logger.info("op=look consumeInfo {}", JsonUtil.obj2String(consumeInfo)); // notify the alert system
             }

src/main/resources/application.yml

@@ -42,7 +42,9 @@ rocketmq:
     # configure multiple namesrv addresses to manage multiple different clusters
     namesrvAddrs:
       - 127.0.0.1:9876
-      - 127.0.0.2:9876
+    #      - 127.0.0.2:9876
+    #      - 10.151.47.32:9876;10.151.47.33:9876;10.151.47.34:9876
+    #      - 10.151.47.30:9876
     # if you use rocketmq version < 3.5.8, rocketmq.config.isVIPChannel should be false.default true
     isVIPChannel:
     # timeout for mqadminExt, default 5000ms
@@ -57,9 +59,12 @@ rocketmq:
     # must create userInfo file: ${rocketmq.config.dataPath}/users.properties if the login is required
     loginRequired: false
     useTLS: false
+    proxyAddr: 127.0.0.1:8080
+    proxyAddrs:
+      - 127.0.0.1:8080
     # set the accessKey and secretKey if you used acl
-    accessKey: # if version > 4.4.0
-    secretKey: # if version > 4.4.0
+#    accessKey: rocketmq2
+#    secretKey: 12345678
 
 threadpool:
   config:

src/main/resources/logback.xml

@@ -25,10 +25,10 @@
 
 	<appender name="FILE"
 		class="ch.qos.logback.core.rolling.RollingFileAppender">
-		<file>${user.home}/logs/consolelogs/rocketmq-console.log</file>
+		<file>${user.home}/logs/dashboardlogs/rocketmq-dashboard.log</file>
 		<append>true</append>
 		<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
-			<fileNamePattern>${user.home}/logs/consolelogs/rocketmq-console-%d{yyyy-MM-dd}.%i.log
+			<fileNamePattern>${user.home}/logs/dashboardlogs/rocketmq-dashboard-%d{yyyy-MM-dd}.%i.log
 			</fileNamePattern>
 			<timeBasedFileNamingAndTriggeringPolicy
 				class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">

src/main/resources/static/index.html

@@ -104,6 +104,7 @@
 <script type="text/javascript" src="src/tools/tools.js?v=201703171710"></script>
 <script type="text/javascript" src="src/cluster.js?timestamp=4"></script>
 <script type="text/javascript" src="src/topic.js"></script>
+<script type="text/javascript" src="src/proxy.js"></script>
 <script type="text/javascript" src="src/consumer.js?timestamp=6"></script>
 <script type="text/javascript" src="src/producer.js"></script>
 <script type="text/javascript" src="src/message.js"></script>

src/main/resources/static/src/app.js

@@ -213,6 +213,9 @@ app.config(['$routeProvider', '$httpProvider','$cookiesProvider','getDictNamePro
         }).when('/ops', {
             templateUrl: 'view/pages/ops.html',
             controller:'opsController'
+        }).when('/proxy', {
+            templateUrl: 'view/pages/proxy.html',
+            controller:'proxyController'
         }).when('/acl', {
             templateUrl: 'view/pages/acl.html',
             controller: 'aclController'

src/main/resources/static/src/consumer.js

@@ -45,6 +45,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
     $scope.writeOperationEnabled =  $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
     $scope.filterNormal = true;
     $scope.filterSystem = false;
+    $scope.filterFIFO = false;
 
     $scope.doSort = function () {// todo  how to change this fe's code ? (it's dirty)
         if ($scope.sortKey == 'diffTotal') {
@@ -75,7 +76,11 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
 
         $http({
             method: "GET",
-            url: "consumer/groupList.query"
+            url: "consumer/groupList.query",
+            params: {
+                skipSysGroup: false,
+                address: localStorage.getItem('isV5') ? localStorage.getItem('proxyAddr') : null
+            }
         }).success(function (resp) {
             if (resp.status == 0) {
                 $scope.allConsumerGrouopList = resp.data;
@@ -135,16 +140,28 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
         $scope.filterList(1);
     });
 
-    $scope.filterByType = function (str) {
+    $scope.$watch('filterFIFO', function () {
+        $scope.filterList(1);
+    });
+
+    $scope.filterByType = function (str, type,version) {
         if ($scope.filterSystem) {
-            if (str.startsWith("%S")) {
+            if (type === "SYSTEM") {
                 return true
             }
         }
         if ($scope.filterNormal) {
-            if (str.startsWith("%") == false) {
+            if (type === "NORMAL") {
                 return true
             }
+            if(!version && type === "FIFO"){
+                return true;
+            }
+        }
+        if ($scope.filterFIFO) {
+            if (type === "FIFO") {
+                return true;
+            }
         }
         return false;
     };
@@ -154,7 +171,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
         var canShowList = [];
         $scope.allConsumerGrouopList.forEach(function (element) {
             console.log(element)
-            if ($scope.filterByType(element.group)) {
+            if ($scope.filterByType(element.group, element.subGroupType, $scope.rmqVersion)) {
                 if (element.group.toLowerCase().indexOf(lowExceptStr) != -1) {
                     canShowList.push(element);
                 }
@@ -189,6 +206,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
                 subscriptionGroupConfig: {
                     groupName: "",
                     consumeEnable: true,
+                    consumeMessageOrderly: false,
                     consumeFromMinEnable: true,
                     consumeBroadcastEnable: true,
                     retryQueueNums: 1,
@@ -211,7 +229,7 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
                         // Refresh topic list
                         $scope.refreshConsumerData();
                     },
-                    template: 'consumerModifyDialog',
+                    template: $scope.rmqVersion ? 'consumerModifyDialogForV5' : 'consumerModifyDialog',
                     controller: 'consumerModifyDialogController',
                     data: {
                         consumerRequestList: request,
@@ -226,11 +244,11 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
             }
         });
     };
-    $scope.detail = function (consumerGroupName) {
+    $scope.detail = function (consumerGroupName, address) {
         $http({
             method: "GET",
             url: "consumer/queryTopicByConsumer.query",
-            params: {consumerGroup: consumerGroupName}
+            params: {consumerGroup: consumerGroupName, address: address}
         }).success(function (resp) {
             if (resp.status == 0) {
                 console.log(resp);
@@ -245,11 +263,11 @@ module.controller('consumerController', ['$scope', 'ngDialog', '$http', 'Notific
         });
     };
 
-    $scope.client = function (consumerGroupName) {
+    $scope.client = function (consumerGroupName, address) {
         $http({
             method: "GET",
             url: "consumer/consumerConnection.query",
-            params: {consumerGroup: consumerGroupName}
+            params: {consumerGroup: consumerGroupName, address: address}
         }).success(function (resp) {
             if (resp.status == 0) {
                 console.log(resp);

src/main/resources/static/src/controller.js

@@ -15,10 +15,18 @@
  * limitations under the License.
  */
 app.controller('AppCtrl', ['$scope','$window','$translate','$http','Notification', function ($scope,$window,$translate, $http, Notification) {
+    $scope.rmqVersion = localStorage.getItem("isV5");
+
     $scope.changeTranslate = function(langKey){
         $translate.use(langKey);
     }
 
+    $scope.changeRMQVersion = function (version) {
+        $scope.rmqVersion = version === 5;
+        var v = version === 5;
+        localStorage.setItem("isV5", v);
+    }
+
     $scope.logout = function(){
         $http({
                     method: "POST",

src/main/resources/static/src/i18n/en.js

@@ -54,9 +54,13 @@ var en = {
     "RESET_CUS_OFFSET": "Reset Consumer Offset",
     "DELETE": "Delete",
     "CHANGE_LANG": "ChangeLanguage",
+    "CHANGE_VERSION": "ChangeVersion",
     "BROKER": "Broker",
     "NORMAL": "NORMAL",
     "RETRY": "RETRY",
+    "FIFO": "FIFO",
+    "TRANSACTION": "TRANSACTION",
+    "UNSPECIFIED": "UNSPECIFIED",
     "DLQ": "DLQ",
     "QUANTITY":"Quantity",
     "TYPE":"Type",
@@ -97,6 +101,7 @@ var en = {
     "RESET_OFFSET":"resetOffset",
     "CLUSTER_NAME":"clusterName",
     "OPS":"OPS",
+    "PROXY":"Proxy",
     "AUTO_REFRESH":"AUTO_REFRESH",
     "REFRESH":"REFRESH",
     "LOGOUT":"Logout",
@@ -123,5 +128,11 @@ var en = {
     "GROUP_PERM":"Group Permission",
     "SYNCHRONIZE":"Synchronize Data",
     "SHOW":"Show",
-    "HIDE":"Hide"
+    "HIDE":"Hide",
+    "MESSAGE_TYPE":"messageType",
+    "MESSAGE_TYPE_UNSPECIFIED": "UNSPECIFIED, is NORMAL",
+    "MESSAGE_TYPE_NORMAL": "NORMAL",
+    "MESSAGE_TYPE_FIFO": "FIFO",
+    "MESSAGE_TYPE_DELAY": "DELAY",
+    "MESSAGE_TYPE_TRANSACTION": "TRANSACTION",
 }

src/main/resources/static/src/i18n/zh.js

@@ -55,9 +55,13 @@ var zh = {
     "SKIP_MESSAGE_ACCUMULATE":"跳过堆积",
     "DELETE": "删除",
     "CHANGE_LANG": "更换语言",
+    "CHANGE_VERSION": "更换版本",
     "BROKER": "Broker",
     "NORMAL": "普通",
     "RETRY": "重试",
+    "FIFO": "顺序",
+    "TRANSACTION": "事务",
+    "UNSPECIFIED": "未指定",
     "DLQ": "死信",
     "QUANTITY":"数量",
     "TYPE":"类型",
@@ -98,6 +102,7 @@ var zh = {
     "RESET_OFFSET":"重置位点",
     "CLUSTER_NAME":"集群名",
     "OPS":"运维",
+    "PROXY":"代理",
     "AUTO_REFRESH":"自动刷新",
     "REFRESH":"刷新",
     "LOGOUT":"退出",
@@ -124,5 +129,11 @@ var zh = {
     "GROUP_PERM":"消费组权限",
     "SYNCHRONIZE":"同步",
     "SHOW":"显示",
-    "HIDE":"隐藏"
+    "HIDE":"隐藏",
+    "MESSAGE_TYPE":"消息类型",
+    "MESSAGE_TYPE_UNSPECIFIED": "未指定,为普通消息",
+    "MESSAGE_TYPE_NORMAL": "普通消息",
+    "MESSAGE_TYPE_FIFO": "顺序消息",
+    "MESSAGE_TYPE_DELAY": "定时/延时消息",
+    "MESSAGE_TYPE_TRANSACTION": "事务消息",
 }

src/main/resources/static/src/proxy.js

@@ -0,0 +1,97 @@
+/*
+ * 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.
+ */
+var module = app;
+module.controller('proxyController', ['$scope', '$location', '$http', 'Notification', 'remoteApi', 'tools', '$window',
+    function ($scope, $location, $http, Notification, remoteApi, tools, $window) {
+        $scope.proxyAddrList = [];
+        $scope.userRole = $window.sessionStorage.getItem("userrole");
+        $scope.writeOperationEnabled = $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
+        $scope.inputReadonly = !$scope.writeOperationEnabled;
+        $scope.newProxyAddr = "";
+        $scope.allProxyConfig = {};
+
+        $http({
+            method: "GET",
+            url: "proxy/homePage.query"
+        }).success(function (resp) {
+            if (resp.status == 0) {
+                $scope.proxyAddrList = resp.data.proxyAddrList;
+                $scope.selectedProxy = resp.data.currentProxyAddr;
+                $scope.showProxyDetailConfig($scope.selectedProxy);
+                localStorage.setItem('proxyAddr',$scope.selectedProxy);
+            } else {
+                Notification.error({message: resp.errMsg, delay: 2000});
+            }
+        });
+
+        $scope.eleChange = function (data) {
+            $scope.proxyAddrList = data;
+        }
+        $scope.showDetailConf = function () {
+            $(".proxyModal").modal();
+        }
+
+
+        $scope.showProxyDetailConfig = function (proxyAddr) {
+            $http({
+                method: "GET",
+                url: "proxy/proxyDetailConfig.query",
+                params: {proxyAddress: proxyAddr}
+            }).success(function (resp) {
+                if (resp.status == 0) {
+                    $scope.allProxyConfig = resp.data;
+                } else {
+                    Notification.error({message: resp.errMsg, delay: 2000});
+                }
+            });
+        };
+
+        $scope.updateProxyAddr = function () {
+            $http({
+                method: "POST",
+                url: "proxy/updateProxyAddr.do",
+                params: {proxyAddr: $scope.selectedProxy}
+            }).success(function (resp) {
+                if (resp.status == 0) {
+                    localStorage.setItem('proxyAddr', $scope.selectedProxy);
+                    Notification.info({message: "SUCCESS", delay: 2000});
+                } else {
+                    Notification.error({message: resp.errMsg, delay: 2000});
+                }
+            });
+            $scope.showProxyDetailConfig($scope.selectedProxy);
+        };
+
+        $scope.addProxyAddr = function () {
+            $http({
+                method: "POST",
+                url: "proxy/addProxyAddr.do",
+                params: {newProxyAddr: $scope.newProxyAddr}
+            }).success(function (resp) {
+                if (resp.status == 0) {
+                    if ($scope.proxyAddrList.indexOf($scope.newProxyAddr) == -1) {
+                        $scope.proxyAddrList.push($scope.newProxyAddr);
+                    }
+                    $("#proxyAddr").val("");
+                    $scope.newProxyAddr = "";
+                    Notification.info({message: "SUCCESS", delay: 2000});
+                } else {
+                    Notification.error({message: resp.errMsg, delay: 2000});
+                }
+            });
+        };
+    }])

src/main/resources/static/src/topic.js

@@ -45,24 +45,31 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
         }
     };
     $scope.filterNormal = true
+    $scope.filterDelay = false
+    $scope.filterFifo = false
+    $scope.filterTransaction = false
+    $scope.filterUnspecified = false
     $scope.filterRetry = false
     $scope.filterDLQ = false
     $scope.filterSystem = false
     $scope.allTopicList = [];
+    $scope.allTopicNameList = [];
+    $scope.allMessageTypeList = [];
     $scope.topicShowList = [];
     $scope.userRole = $window.sessionStorage.getItem("userrole");
-    $scope.writeOperationEnabled =  $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
+    $scope.writeOperationEnabled = $scope.userRole == null ? true : ($scope.userRole == 1 ? true : false);
 
     $scope.refreshTopicList = function () {
         $http({
             method: "GET",
-            url: "topic/list.query"
+            url: "topic/list.queryTopicType"
         }).success(function (resp) {
             if (resp.status == 0) {
-                $scope.allTopicList = resp.data.topicList.sort();
-                console.log($scope.allTopicList);
+                $scope.allTopicNameList = resp.data.topicNameList;
+                $scope.allMessageTypeList = resp.data.messageTypeList;
+                console.log($scope.allTopicNameList);
                 console.log(JSON.stringify(resp));
-                $scope.showTopicList(1, $scope.allTopicList.length);
+                $scope.showTopicList(1, $scope.allTopicNameList.length);
 
             } else {
                 Notification.error({message: resp.errMsg, delay: 5000});
@@ -79,6 +86,18 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
     $scope.$watch('filterNormal', function () {
         $scope.filterList(1);
     });
+    $scope.$watch('filterFifo', function () {
+        $scope.filterList(1);
+    });
+    $scope.$watch('filterTransaction', function () {
+        $scope.filterList(1);
+    });
+    $scope.$watch('filterUnspecified', function () {
+        $scope.filterList(1);
+    });
+    $scope.$watch('filterDelay', function () {
+        $scope.filterList(1);
+    });
     $scope.$watch('filterRetry', function () {
         $scope.filterList(1);
     });
@@ -92,13 +111,13 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
         var lowExceptStr = $scope.filterStr.toLowerCase();
         var canShowList = [];
 
-        $scope.allTopicList.forEach(function (element) {
-            if ($scope.filterByType(element)) {
-                if (element.toLowerCase().indexOf(lowExceptStr) != -1) {
-                    canShowList.push(element);
+        for (let i = 0; i < $scope.allTopicNameList.length; ++i) {
+            if ($scope.filterByType($scope.allTopicNameList[i], $scope.allMessageTypeList[i])) {
+                if ($scope.allTopicNameList[i].toLowerCase().indexOf(lowExceptStr) != -1) {
+                    canShowList.push($scope.allTopicNameList[i]);
                 }
             }
-        });
+        }
         $scope.paginationConf.totalItems = canShowList.length;
         var perPage = $scope.paginationConf.itemsPerPage;
         var from = (currentPage - 1) * perPage;
@@ -106,7 +125,7 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
         $scope.topicShowList = canShowList.slice(from, to);
     };
 
-    $scope.filterByType = function (str) {
+    $scope.filterByType = function (str, type) {
         if ($scope.filterRetry) {
             if (str.startsWith("%R")) {
                 return true
@@ -122,8 +141,31 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
                 return true
             }
         }
+        if (localStorage.getItem('isV5') && $scope.filterUnspecified) {
+            if (type.includes("UNSPECIFIED")) {
+                return true
+            }
+        }
         if ($scope.filterNormal) {
-            if (str.startsWith("%") == false) {
+            if (type.includes("NORMAL")) {
+                return true
+            }
+            if (!localStorage.getItem('isV5') && type.includes("UNSPECIFIED")) {
+                return true
+            }
+        }
+        if (localStorage.getItem('isV5') && $scope.filterDelay) {
+            if (type.includes("DELAY")) {
+                return true
+            }
+        }
+        if (localStorage.getItem('isV5') && $scope.filterFifo) {
+            if (type.includes("FIFO")) {
+                return true
+            }
+        }
+        if (localStorage.getItem('isV5') && $scope.filterTransaction) {
+            if (type.includes("TRANSACTION")) {
                 return true
             }
         }
@@ -138,10 +180,10 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
         var perPage = $scope.paginationConf.itemsPerPage;
         var from = (currentPage - 1) * perPage;
         var to = (from + perPage) > totalItem ? totalItem : from + perPage;
-        console.log($scope.allTopicList);
+        console.log($scope.allTopicNameList);
         console.log(from)
         console.log(to)
-        $scope.topicShowList = $scope.allTopicList.slice(from, to);
+        $scope.topicShowList = $scope.allTopicNameList.slice(from, to);
         $scope.paginationConf.totalItems = totalItem;
         console.log($scope.topicShowList)
         console.log($scope.paginationConf.totalItems)
@@ -328,8 +370,8 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
         var bIsUpdate = true;
         if (request == null) {
             request = [{
-                writeQueueNums: 16,
-                readQueueNums: 16,
+                writeQueueNums: 8,
+                readQueueNums: 8,
                 perm: 6,
                 order: false,
                 topicName: "",
@@ -355,6 +397,7 @@ module.controller('topicController', ['$scope', 'ngDialog', '$http', 'Notificati
                         topicRequestList: request,
                         allClusterNameList: Object.keys(resp.data.clusterInfo.clusterAddrTable),
                         allBrokerNameList: Object.keys(resp.data.brokerServer),
+                        allMessageTypeList: resp.data.messageTypes,
                         bIsUpdate: bIsUpdate,
                         writeOperationEnabled: $scope.writeOperationEnabled
                     }

src/main/resources/static/view/layout/_header.html

@@ -28,6 +28,7 @@
         <div class="navbar-collapse collapse navbar-warning-collapse">
             <ul class="nav navbar-nav">
                 <li ng-class="path =='ops' ? 'active':''"><a ng-href="#/ops">{{'OPS' | translate}}</a></li>
+                <li ng-show="rmqVersion" ng-class="path =='proxy' ? 'active':''"><a ng-href="#/proxy">{{'PROXY' | translate}}</a></li>
                 <li ng-class="path =='dashboard' || path ==''? 'active':''"><a ng-href="#/">{{'DASHBOARD' | translate}}</a></li>
                 <li ng-class="path =='cluster' ? 'active':''"><a ng-href="#/cluster">{{'CLUSTER' | translate}}</a></li>
                 <li ng-class="path =='topic' ? 'active':''"><a ng-href="#/topic">{{'TOPIC' | translate}}</a></li>
@@ -47,6 +48,14 @@
                         <li><a href="javascript:void(0)"  ng-click="changeTranslate('zh')">Simplified Chinese</a></li>
                     </ul>
                 </li>
+                <li class="dropdown">
+                    <a href="bootstrap-elements.html" data-target="#" class="dropdown-toggle" data-toggle="dropdown">{{'CHANGE_VERSION' | translate}}
+                        <b class="caret"></b></a>
+                    <ul class="dropdown-menu">
+                        <li><a href="javascript:void(0)" ng-click="changeRMQVersion(5)">RocketMQ 5.x</a></li>
+                        <li><a href="javascript:void(0)"  ng-click="changeRMQVersion(4)">RocketMQ 4.x</a></li>
+                    </ul>
+                </li>
                 <li class="dropdown" ng-show="username != ''">
                     <a href="bootstrap-elements.html" data-target="#" class="dropdown-toggle" data-toggle="dropdown">{{username}}
                         <b class="caret"></b></a>

src/main/resources/static/view/pages/cluster.html

@@ -50,7 +50,14 @@
                 <td class="text-center">{{instance.address}}</td>
                 <td class="text-center">{{instance.brokerVersionDesc}}</td>
                 <td class="text-center">{{instance.putTps.split(' ')[0]| number:2}}</td>
-                <td class="text-center">{{instance.getTransferedTps.split(' ')[0]| number:2}}</td>
+                <td class="text-center">
+                    <span ng-if="!instance.getTransferedTps || !instance.getTransferedTps.trim()">
+                        {{instance.getTransferredTps.split(' ')[0] | number:2}}
+                    </span>
+                    <span ng-if="instance.getTransferedTps && instance.getTransferedTps.trim()">
+                        {{instance.getTransferedTps.split(' ')[0] | number:2}}
+                    </span>
+                </td>
                 <td class="text-center">{{instance.msgPutTotalTodayMorning -
                     instance.msgPutTotalYesterdayMorning}}
                 </td>

src/main/resources/static/view/pages/consumer.html

@@ -26,6 +26,8 @@
                 </div>
                 <md-checkbox aria-label="Checkbox" ng-model="filterNormal" class="md-primary">{{'NORMAL' | translate}}
                 </md-checkbox>
+                <md-checkbox aria-label="Checkbox" ng-show="rmqVersion" ng-model="filterFIFO" class="md-primary">{{'FIFO' | translate}}
+                </md-checkbox>
                 <md-checkbox aria-label="Checkbox" ng-model="filterSystem" class="md-primary">{{'SYSTEM' | translate}}
                 </md-checkbox>
                 <button class="btn btn-raised btn-sm btn-primary" type="button" ng-show="{{writeOperationEnabled}}"
@@ -64,11 +66,11 @@
                         <td class="text-center">{{consumerGroup.consumeTps}}</td>
                         <td class="text-center">{{consumerGroup.diffTotal}}</td>
                         <td class="text-left">
-                            <button name="client" ng-click="client(consumerGroup.group)"
+                            <button name="client" ng-click="client(consumerGroup.group, consumerGroup.address)"
                                     class="btn btn-raised btn-sm btn-primary"
                                     type="button">{{'CLIENT' | translate}}
                             </button>
-                            <button name="client" ng-click="detail(consumerGroup.group)"
+                            <button name="client" ng-click="detail(consumerGroup.group, consumerGroup.address)"
                                     class="btn btn-raised btn-sm btn-primary"
                                     type="button">{{'CONSUME_DETAIL' | translate}}
                             </button>
@@ -274,15 +276,6 @@
                             <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
                         </div>
                     </div>
-                    <!--<div class="form-group">-->
-                    <!--<label class="control-label col-sm-4">retryMaxTimes:</label>-->
-                    <!--<div class="col-sm-8">-->
-                    <!--<input class="form-control" ng-model="item.subscriptionGroupConfig.retryMaxTimes"-->
-                    <!--type="text"-->
-                    <!--required/>-->
-                    <!--<span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>-->
-                    <!--</div>-->
-                    <!--</div>-->
                     <div class="form-group">
                         <label class="control-label col-sm-3">brokerId:</label>
                         <div class="col-sm-9">
@@ -291,6 +284,132 @@
                             <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
                         </div>
                     </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">retryMaxTimes:</label>
+                        <div class="col-sm-9">
+                            <input class="form-control" ng-model="item.subscriptionGroupConfig.retryMaxTimes" type="text"
+                                   ng-disabled="{{!ngDialogData.writeOperationEnabled}}" required/>
+                            <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">whichBrokerWhenConsumeSlowly:</label>
+                        <div class="col-sm-9">
+                            <input class="form-control"
+                                   ng-model="item.subscriptionGroupConfig.whichBrokerWhenConsumeSlowly" type="text"
+                                   ng-disabled="{{!ngDialogData.writeOperationEnabled}}" required/>
+                            <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
+                        </div>
+                    </div>
+                </form>
+                <div class="modal-footer">
+                    <div class="ngdialog-buttons">
+                        <button type="button" class="ngdialog-button ngdialog-button-primary"
+                                ng-disabled="addAppForm.$invalid"
+                                ng-show="{{ngDialogData.writeOperationEnabled}}"
+                                ng-click="postConsumerRequest(item)">{{ 'COMMIT' | translate }}
+                        </button>
+                        <button type="button" class="ngdialog-button ngdialog-button-secondary"
+                                ng-click="closeThisDialog('Cancel')">{{ 'CLOSE' | translate }}
+                        </button>
+                    </div>
+                </div>
+            </div>
+        </div>
+    </div>
+
+    </div>
+</script>
+
+<script type="text/ng-template" id="consumerModifyDialogForV5">
+    <div>
+        <div>
+            <div class="modal-header">
+                <h4 class="modal-title">{{'SUBSCRIPTION_CHANGE'|translate}}</h4>
+            </div>
+            <div class="modal-body " ng-repeat="item in ngDialogData.consumerRequestList">
+                <form id="addAppForm1" name="addAppForm" class="form-horizontal" novalidate>
+                    <div class="form-group" ng-hide="ngDialogData.bIsUpdate">
+                        <label class="control-label col-sm-3">clusterName:</label>
+                        <div class="col-sm-9">
+                            <select name="mySelectClusterNameList" multiple chosen
+                                    ng-model="item.clusterNameList"
+                                    ng-options="clusterNameItem for clusterNameItem in ngDialogData.allClusterNameList">
+                                <option value=""></option>
+                            </select>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">brokerName:</label>
+                        <div class="col-sm-9">
+                            <select name="mySelectBrokerNameList" multiple chosen
+                                    ng-disabled="ngDialogData.bIsUpdate"
+                                    ng-model="item.brokerNameList"
+                                    ng-options="brokerNameItem for brokerNameItem in ngDialogData.allBrokerNameList">
+                                <option value=""></option>
+                            </select>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">groupName:</label>
+                        <div class="col-sm-9">
+                            <input class="form-control" ng-model="item.subscriptionGroupConfig.groupName" type="text"
+                                   ng-disabled="ngDialogData.bIsUpdate" required/>
+                            <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">consumeEnable:</label>
+                        <div class="col-sm-9">
+                            <md-switch class="md-primary" ng-disabled="{{!ngDialogData.writeOperationEnabled}}" md-no-ink
+                                       aria-label="Switch No Ink" ng-model="item.subscriptionGroupConfig.consumeEnable">
+                            </md-switch>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">consumeOrderlyEnable:</label>
+                        <div class="col-sm-9">
+                            <md-switch class="md-primary custom-md-switch" eng-disabled="{{!ngDialogData.writeOperationEnabled}}" md-no-ink
+                                       aria-label="Switch No Ink" ng-model="item.subscriptionGroupConfig.consumeMessageOrderly">
+                            </md-switch>
+                            <span style="font-size: 12px;">[Pay Attention: FIFO ConsumerGroup Need Open 'consumeOrderlyEnable' Option]</span>
+                        </div>
+
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">consumeBroadcastEnable:</label>
+                        <div class="col-sm-9">
+                            <md-switch class="md-primary" ng-disabled="{{!ngDialogData.writeOperationEnabled}}" md-no-ink
+                                       aria-label="Switch No Ink"
+                                       ng-model="item.subscriptionGroupConfig.consumeBroadcastEnable">
+                            </md-switch>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">retryQueueNums:</label>
+                        <div class="col-sm-9">
+                            <input class="form-control" ng-model="item.subscriptionGroupConfig.retryQueueNums"
+                                   type="text" ng-disabled="{{!ngDialogData.writeOperationEnabled}}"
+                                   required/>
+                            <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">brokerId:</label>
+                        <div class="col-sm-9">
+                            <input class="form-control" ng-model="item.subscriptionGroupConfig.brokerId" type="text"
+                                   ng-disabled="{{!ngDialogData.writeOperationEnabled}}" required/>
+                            <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
+                        </div>
+                    </div>
+                    <div class="form-group">
+                        <label class="control-label col-sm-3">retryMaxTimes:</label>
+                        <div class="col-sm-9">
+                            <input class="form-control" ng-model="item.subscriptionGroupConfig.retryMaxTimes" type="text"
+                                   ng-disabled="{{!ngDialogData.writeOperationEnabled}}" required/>
+                            <span class="text-danger" ng-show="addAppForm.name.$error.required">编号不能为空.</span>
+                        </div>
+                    </div>
                     <div class="form-group">
                         <label class="control-label col-sm-3">whichBrokerWhenConsumeSlowly:</label>
                         <div class="col-sm-9">

src/main/resources/static/view/pages/proxy.html

@@ -0,0 +1,67 @@
+<!--
+  ~ 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.
+  -->
+<div class="container-fluid" id="deployHistoryList">
+    <div class="page-content">
+        <h2 class="md-title">ProxyServerAddressList</h2>
+        <div class="pull-left" style="min-width: 400px; max-width: 500px; padding: 10px 10px 10px 0">
+            <select ng-model="selectedProxy" chosen
+                    ng-options="x for x in proxyAddrList"
+                    ng-change="updateProxyAddr()"
+                    required></select>
+        </div>
+        <div class="pull-left">
+            <button class="btn btn-raised btn-sm btn-primary" type="button" ng-show="{{writeOperationEnabled}}"
+                    ng-click="updateProxyAddr()">{{'UPDATE' | translate}}
+            </button>
+        </div>
+        <form class="form-inline pull-left" style="margin-left: 20px" ng-show="{{writeOperationEnabled}}">
+            <div class="form-group" style="margin: 0">
+                <label for="proxyAddr">ProxyAddr:</label>
+                <input id="proxyAddr" class="form-control" style="width: 300px; margin: 0 10px 0 10px" type="text" ng-model="newProxyAddr" required/>
+                <button class="btn btn-raised btn-sm btn-primary" type="button"
+                        ng-click="addProxyAddr()"> {{ 'ADD' | translate}}
+                </button>
+            </div>
+        </form>
+    </div>
+</div>
+
+<div class="modal proxyModal fade" role="dialog" tabindex="-1" aria-hidden="true" aria-labelledby="config-modal-label">
+    <div class="modal-dialog modal-lg">
+        <div class="modal-content" >
+            <div class="modal-header">
+                <button class="close" type="button" data-dismiss="modal">&times;</button>
+                <h4 id="config-modal-label" class="modal-title">
+                    [{{selectedProxy}}]
+                </h4>
+            </div>
+            <div class="modal-body limit_height">
+                <table class="table table-bordered">
+                    <tr ng-repeat="(key, value) in allProxyConfig">
+                        <td>{{key}}</td>
+                        <td>{{value}}</td>
+                    </tr>
+                </table>
+            </div>
+            <div class="modal-footer">
+                <div class="col-md-12 text-center">
+                    <button type="button" class="btn btn-raised" data-dismiss="modal">{{ 'CLOSE' | translate }}</button>
+                </div>
+            </div>
+        </div>
+    </div>
+</div>

src/main/resources/static/view/pages/topic.html

@@ -24,6 +24,14 @@
                 </div>
                 <md-checkbox aria-label="Checkbox" ng-model="filterNormal" class="md-primary">{{'NORMAL' | translate}}
                 </md-checkbox>
+                <md-checkbox aria-label="Checkbox" ng-model="filterDelay" class="md-primary" ng-show="rmqVersion">{{'DELAY' | translate}}
+                </md-checkbox>
+                <md-checkbox aria-label="Checkbox" ng-model="filterFifo" class="md-primary" ng-show="rmqVersion">{{'FIFO' | translate}}
+                </md-checkbox>
+                <md-checkbox aria-label="Checkbox" ng-model="filterTransaction" class="md-primary" ng-show="rmqVersion">{{'TRANSACTION' | translate}}
+                </md-checkbox>
+                <md-checkbox aria-label="Checkbox" ng-model="filterUnspecified" class="md-primary" ng-show="rmqVersion">{{'UNSPECIFIED' | translate}}
+                </md-checkbox>
                 <md-checkbox aria-label="Checkbox" ng-model="filterRetry" class="md-primary">{{'RETRY' | translate}}
                 </md-checkbox>
                 <md-checkbox aria-label="Checkbox" ng-model="filterDLQ" class="md-primary">{{'DLQ' | translate}}
@@ -63,6 +71,7 @@
                             <button class="btn btn-raised btn-sm btn-primary" type="button"
                                     ng-click="openUpdateDialog(topic, sysFlag)">topic {{'CONFIG' |translate}}
                             </button>
+<!--                            todo 发送消息，根据消息类型判断-->
                             <button class="btn btn-raised btn-sm btn-primary" type="button"
                                     ng-show="{{!sysFlag}}"
                                     ng-click="openSendTopicMessageDialog(topic)">{{'SEND_MSG' | translate}}
@@ -189,6 +198,18 @@
                     <span class="text-danger" ng-show="addAppForm.topicName.$error.required">{{'TOPIC_NAME'|translate}}不能为空.</span>
                 </div>
             </div>
+            <!--            设置topic 类型 -->
+            <div class="form-group">
+                <label class="control-label col-sm-2">{{'MESSAGE_TYPE'|translate}}:</label>
+                <div class="col-sm-10">
+                    <select name="mySelectMessageType" chosen ng-disabled="ngDialogData.bIsUpdate"
+                            ng-model="item.messageType"
+                            ng-options="messageType as value | translate  disable when messageType=='UNSPECIFIED' for (messageType , value) in ngDialogData.allMessageTypeList"
+                    >
+                        <option value=""></option>
+                    </select>
+                </div>
+            </div>
             <div class="form-group">
                 <label class="control-label col-sm-2">{{'WRITE_QUEUE_NUMS'|translate}}:</label>
                 <div class="col-sm-10">

src/test/java/org/apache/rocketmq/dashboard/admin/MQAdminExtImplTest.java

@@ -34,27 +34,27 @@ import org.apache.rocketmq.client.impl.MQClientAPIImpl;
 import org.apache.rocketmq.client.impl.factory.MQClientInstance;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
-import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.service.client.MQAdminExtImpl;
 import org.apache.rocketmq.dashboard.service.client.MQAdminInstance;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
@@ -82,6 +82,7 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
@@ -570,11 +571,13 @@ public class MQAdminExtImplTest {
     public void testCreateTopic() throws Exception {
         assertNotNull(mqAdminExtImpl);
         {
-            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt());
-            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyInt());
+            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyMap());
+            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyInt(), anyMap());
         }
-        mqAdminExtImpl.createTopic("key", "topic_test", 8);
-        mqAdminExtImpl.createTopic("key", "topic_test", 8, 1);
+        Map<String, String> map = new HashMap<>();
+        map.put("message.type", "FIFO");
+        mqAdminExtImpl.createTopic("key", "topic_test", 8, map);
+        mqAdminExtImpl.createTopic("key", "topic_test", 8, 1, map);
     }
 
     @Test

src/test/java/org/apache/rocketmq/dashboard/admin/MQAdminPoolTest.java

@@ -18,7 +18,7 @@ package org.apache.rocketmq.dashboard.admin;
 
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.GenericObjectPool;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
 import org.apache.rocketmq.tools.admin.MQAdminExt;

src/test/java/org/apache/rocketmq/dashboard/controller/AclControllerTest.java

@@ -21,7 +21,7 @@ import com.google.common.collect.Lists;
 import java.util.List;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.dashboard.model.request.AclRequest;
 import org.apache.rocketmq.dashboard.service.impl.AclServiceImpl;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;

src/test/java/org/apache/rocketmq/dashboard/controller/ClusterControllerTest.java

@@ -18,8 +18,8 @@ package org.apache.rocketmq.dashboard.controller;
 
 import java.util.HashMap;
 import java.util.Properties;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
 import org.apache.rocketmq.dashboard.service.impl.ClusterServiceImpl;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
 import org.junit.Test;

src/test/java/org/apache/rocketmq/dashboard/controller/ConsumerControllerTest.java

@@ -23,17 +23,17 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
-import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
+import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
 import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
 import org.apache.rocketmq.dashboard.model.request.ResetOffsetRequest;
@@ -67,6 +67,7 @@ public class ConsumerControllerTest extends BaseControllerTest {
 
     @Before
     public void init() throws Exception {
+        consumerService.afterPropertiesSet();
         super.mockRmqConfigure();
         ClusterInfo clusterInfo = MockObjectUtil.createClusterInfo();
         when(mqAdminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo);
@@ -93,9 +94,10 @@ public class ConsumerControllerTest extends BaseControllerTest {
         perform = mockMvc.perform(requestBuilder);
         perform.andExpect(status().isOk())
             .andExpect(jsonPath("$.data", hasSize(2)))
-            .andExpect(jsonPath("$.data[0].group").value("group_test"))
             .andExpect(jsonPath("$.data[0].consumeType").value(ConsumeType.CONSUME_ACTIVELY.name()))
             .andExpect(jsonPath("$.data[0].messageModel").value(MessageModel.CLUSTERING.name()));
+        // executorService shutdown
+        consumerService.destroy();
     }
 
     @Test

src/test/java/org/apache/rocketmq/dashboard/controller/DlqMessageControllerTest.java

@@ -21,10 +21,10 @@ import com.google.common.collect.Lists;
 import java.util.List;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.CMResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.CMResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
 import org.apache.rocketmq.dashboard.model.MessagePage;
 import org.apache.rocketmq.dashboard.model.MessageView;

src/test/java/org/apache/rocketmq/dashboard/controller/MessageControllerTest.java

@@ -32,9 +32,9 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.CMResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.CMResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
 import org.apache.rocketmq.dashboard.model.request.MessageQuery;
 import org.apache.rocketmq.dashboard.service.impl.MessageServiceImpl;

src/test/java/org/apache/rocketmq/dashboard/controller/ProducerControllerTest.java

@@ -19,8 +19,8 @@ package org.apache.rocketmq.dashboard.controller;
 
 import java.util.HashSet;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
 import org.apache.rocketmq.dashboard.interceptor.AuthInterceptor;
 import org.apache.rocketmq.dashboard.service.impl.LoginServiceImpl;
 import org.apache.rocketmq.dashboard.service.impl.ProducerServiceImpl;

src/test/java/org/apache/rocketmq/dashboard/controller/TopicControllerTest.java

@@ -29,16 +29,16 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
 import org.apache.rocketmq.dashboard.service.impl.ConsumerServiceImpl;
@@ -54,6 +54,7 @@ import org.springframework.http.MediaType;
 import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
 
 import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -135,8 +136,9 @@ public class TopicControllerTest extends BaseControllerTest {
         // 3、filter system topic
         requestBuilder = MockMvcRequestBuilders.get(url);
         perform = mockMvc.perform(requestBuilder);
+        String[] topicString = {"%SYS%system_topic2","common_topic2","%SYS%system_topic1","common_topic1"};
         perform.andExpect(status().isOk())
-            .andExpect(jsonPath("$.data.topicList[2]").value("%SYS%system_topic1"));
+            .andExpect(jsonPath("$.data.topicList").value(containsInAnyOrder(topicString)));
     }
 
     @Test

src/test/java/org/apache/rocketmq/dashboard/task/DashboardCollectTaskTest.java

@@ -33,12 +33,12 @@ import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.BaseTest;
 import org.apache.rocketmq.dashboard.config.CollectExecutorConfig;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;

src/test/java/org/apache/rocketmq/dashboard/testbase/RocketMQConsoleTestBase.java

@@ -29,7 +29,7 @@ import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
 import org.apache.rocketmq.dashboard.service.ConsumerService;
@@ -78,7 +78,8 @@ public abstract class RocketMQConsoleTestBase {
                     }
                 }
             }
-            throw Throwables.propagate(exception);
+            Throwables.throwIfUnchecked(exception);
+            throw new RuntimeException(exception);
         }
     }
 
@@ -91,7 +92,8 @@ public abstract class RocketMQConsoleTestBase {
             producer.start();
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 
@@ -137,7 +139,8 @@ public abstract class RocketMQConsoleTestBase {
             consumer.start();
         }
         catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
         }
     }
 

src/test/java/org/apache/rocketmq/dashboard/util/MockObjectUtil.java

@@ -33,39 +33,39 @@ import org.apache.rocketmq.client.producer.LocalTransactionState;
 import org.apache.rocketmq.client.trace.TraceConstants;
 import org.apache.rocketmq.client.trace.TraceType;
 import org.apache.rocketmq.common.AclConfig;
-import org.apache.rocketmq.common.DataVersion;
+import org.apache.rocketmq.remoting.protocol.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.OffsetWrapper;
-import org.apache.rocketmq.common.admin.TopicOffset;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsItem;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ConsumeStatus;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.ProcessQueueInfo;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
-import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
-import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
-import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.protocol.route.QueueData;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsItem;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeStatus;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.ProcessQueueInfo;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
+import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
 import org.apache.rocketmq.remoting.protocol.LanguageCode;
 import org.checkerframework.checker.units.qual.A;
 
-import static org.apache.rocketmq.common.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY;
+import static org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY;
 
 public class MockObjectUtil {