Merge a172446767 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>

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": "5.0.1",
+    "react-scripts": "4.0.3",
     "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>
@@ -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/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

@@ -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/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/GroupConsumeInfo.java

@@ -19,14 +19,19 @@ package org.apache.rocketmq.dashboard.model;
 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;
     private String version;
     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/request/TopicConfigInfo.java

@@ -31,6 +31,7 @@ public class TopicConfigInfo {
     private int perm;
     private boolean order;
 
+    private String messageType;
     public List<String> getClusterNameList() {
         return clusterNameList;
     }
@@ -91,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)
@@ -102,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/ConsumerService.java

@@ -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/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,6 +19,7 @@ package org.apache.rocketmq.dashboard.service;
 
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.TopicConfig;
+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;
@@ -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,27 +17,29 @@
 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.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.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
 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;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
+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;
@@ -64,23 +66,20 @@ 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.dashboard.util.JsonUtil;
-import org.apache.rocketmq.remoting.RemotingClient;
-import org.apache.rocketmq.remoting.exception.RemotingCommandException;
-import org.apache.rocketmq.remoting.exception.RemotingConnectException;
-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.tools.admin.MQAdminExt;
 import org.apache.rocketmq.tools.admin.api.BrokerOperatorResult;
 import org.apache.rocketmq.tools.admin.api.MessageTrack;
 import org.apache.rocketmq.tools.admin.common.AdminToolResult;
-import org.joor.Reflect;
 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
@@ -465,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
@@ -627,7 +629,7 @@ public class MQAdminExtImpl implements MQAdminExt {
             long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
             RemotingConnectException, MQBrokerException {
         // TODO Auto-generated method stub
-        throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStats'");
+        return MQAdminInstance.threadLocalMQAdminExt().examineConsumeStats(brokerAddr, consumerGroup, topicName, timeoutMillis);
     }
 
     @Override
@@ -639,8 +641,7 @@ public class MQAdminExtImpl implements MQAdminExt {
     @Override
     public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup, String brokerAddr)
             throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
-        // TODO Auto-generated method stub
-        throw new UnsupportedOperationException("Unimplemented method 'examineConsumerConnectionInfo'");
+        return MQAdminInstance.threadLocalMQAdminExt().examineConsumerConnectionInfo(consumerGroup, brokerAddr);
     }
 
     @Override

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/ClusterServiceImpl.java

@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.dashboard.service.impl;
 
+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;
@@ -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,6 +59,9 @@ 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) {

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

@@ -23,8 +23,10 @@ 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;
@@ -44,6 +46,7 @@ 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.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;
@@ -77,6 +80,8 @@ import org.springframework.stereotype.Service;
 public class ConsumerServiceImpl extends AbstractCommonService implements ConsumerService, InitializingBean, DisposableBean {
     private Logger logger = LoggerFactory.getLogger(ConsumerServiceImpl.class);
 
+    @Resource
+    protected ProxyAdmin proxyAdmin;
     @Resource
     private RMQConfigure configure;
 
@@ -119,25 +124,33 @@ 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) {
+        } catch (Exception err) {
             Throwables.throwIfUnchecked(err);
             throw new RuntimeException(err);
         }
         List<GroupConsumeInfo> groupConsumeInfoList = Collections.synchronizedList(Lists.newArrayList());
-        CountDownLatch countDownLatch = new CountDownLatch(consumerGroupSet.size());
-        for (String consumerGroup : consumerGroupSet) {
+        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);
+                    GroupConsumeInfo consumeInfo = queryGroup(consumerGroup, address);
+                    consumeInfo.setAddress(entry.getValue());
                     groupConsumeInfoList.add(consumeInfo);
                 } catch (Exception e) {
                     logger.error("queryGroup exception, consumerGroup: {}", consumerGroup, e);
@@ -165,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;
@@ -177,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());
@@ -206,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
@@ -220,6 +257,10 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
             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) {
@@ -420,11 +461,12 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
     }
 
     @Override
-    public ConsumerConnection getConsumerConnection(String consumerGroup) {
+    public ConsumerConnection getConsumerConnection(String consumerGroup, String address) {
         try {
-            return mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
-        }
-        catch (Exception e) {
+            String[] addresses = address.split(",");
+            String addr = addresses[0];
+            return mqAdminExt.examineConsumerConnectionInfo(consumerGroup, addr);
+        } catch (Exception e) {
             Throwables.throwIfUnchecked(e);
             throw new RuntimeException(e);
         }

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

@@ -276,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));
             }
 

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.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
 import org.apache.rocketmq.common.message.Message;
+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.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.service.AbstractCommonService;
-import org.apache.rocketmq.dashboard.service.TopicService;
-import org.apache.rocketmq.remoting.RPCHook;
 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,18 +82,18 @@ 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;
@@ -89,6 +103,41 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
         }
     }
 
+    @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);
+        }
+    }
+
     @Override
     public TopicStatsTable stats(String topic) {
         try {
@@ -123,10 +172,15 @@ 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) {
@@ -156,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;
@@ -226,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());
@@ -249,32 +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) {
-            Throwables.throwIfUnchecked(e);
-            throw new RuntimeException(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) {
@@ -296,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.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) {

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/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">