Merge a4a6000734 into 823bce2b8b

feat: add topic message type
add message type
2026-06-07 04:23:20 +08:00 · 2024-06-04 15:03:02 +08:00 · 2024-06-04 11:40:46 +08:00 · 2024-03-26 17:02:16 +08:00 · 2023-04-19 20:33:17 +08:00 · 2023-04-07 08:30:18 +08:00
49 changed files with 739 additions and 283 deletions
--- a/.dockerignore
+++ b/.dockerignore
@@ -0,0 +1 @@
+target/
--- a/pom.xml
+++ b/pom.xml
@@ -91,7 +91,7 @@
        <commons-io.version>2.4</commons-io.version>
        <commons-cli.version>1.2</commons-cli.version>
        <commons-collections.version>3.2.2</commons-collections.version>
-        <rocketmq.version>4.9.3</rocketmq.version>
+        <rocketmq.version>5.1.0</rocketmq.version>
        <surefire.version>2.19.1</surefire.version>
        <aspectj.version>1.9.6</aspectj.version>
        <lombok.version>1.18.22</lombok.version>
--- a/src/main/docker/Dockerfile
+++ b/src/main/docker/Dockerfile
@@ -15,9 +15,16 @@
 # limitations under the License.
 #

-FROM java:8
-VOLUME /tmp
-ADD rocketmq-dashboard-*.jar rocketmq-dashboard.jar
-RUN sh -c 'touch /rocketmq-dashboard.jar'
-ENV JAVA_OPTS=""
-ENTRYPOINT [ "sh", "-c", "java $JAVA_OPTS -jar /rocketmq-dashboard.jar" ]
+FROM maven:3.8.6-openjdk-8 AS builder
+
+ADD . .
+
+# package jar
+RUN mvn clean package -Dmaven.test.skip=true
+
+FROM openjdk:8u342-jdk
+
+# copy jar from the builder stage
+COPY --from=builder target/rocketmq-dashboard-*.jar rocketmq-dashboard.jar
+
+ENTRYPOINT exec java $JAVA_OPTS -jar rocketmq-dashboard.jar
--- a/src/main/java/org/apache/rocketmq/dashboard/admin/MQAdminPooledObjectFactory.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/admin/MQAdminPooledObjectFactory.java
@@ -21,7 +21,7 @@ import org.apache.commons.collections.MapUtils;
 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.PooledObjectFactory;
 import org.apache.commons.pool2.impl.DefaultPooledObject;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.tools.admin.MQAdminExt;

@Slf4j
--- a/src/main/java/org/apache/rocketmq/dashboard/controller/ConsumerController.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/controller/ConsumerController.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.controller;
 import com.google.common.base.Preconditions;
 import javax.annotation.Resource;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.dashboard.model.ConnectionInfo;
 import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
 import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
--- a/src/main/java/org/apache/rocketmq/dashboard/controller/MessageController.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/controller/MessageController.java
@@ -18,7 +18,7 @@ package org.apache.rocketmq.dashboard.controller;

 import com.google.common.collect.Maps;
 import org.apache.rocketmq.common.Pair;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 import org.apache.rocketmq.dashboard.model.MessagePage;
 import org.apache.rocketmq.dashboard.model.MessageView;
 import org.apache.rocketmq.dashboard.model.request.MessageQuery;
--- a/src/main/java/org/apache/rocketmq/dashboard/controller/ProducerController.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/controller/ProducerController.java
@@ -17,7 +17,7 @@
 package org.apache.rocketmq.dashboard.controller;

 import javax.annotation.Resource;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
 import org.apache.rocketmq.dashboard.model.ConnectionInfo;
 import org.apache.rocketmq.dashboard.permisssion.Permission;
 import org.apache.rocketmq.dashboard.service.ProducerService;
--- a/src/main/java/org/apache/rocketmq/dashboard/model/ConnectionInfo.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/ConnectionInfo.java
@@ -20,7 +20,7 @@ import com.google.common.collect.Sets;
 import java.util.Collection;
 import java.util.HashSet;
 import org.apache.rocketmq.common.MQVersion;
-import org.apache.rocketmq.common.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.Connection;

 public class ConnectionInfo extends Connection {
    private String versionDesc;
--- a/src/main/java/org/apache/rocketmq/dashboard/model/ConsumerGroupRollBackStat.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/ConsumerGroupRollBackStat.java
@@ -16,7 +16,7 @@
 */
 package org.apache.rocketmq.dashboard.model;

-import org.apache.rocketmq.common.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
 import com.google.common.collect.Lists;

 import java.util.List;
--- a/src/main/java/org/apache/rocketmq/dashboard/model/DlqMessageResendResult.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/DlqMessageResendResult.java
@@ -17,8 +17,8 @@
 package org.apache.rocketmq.dashboard.model;

 import lombok.Data;
-import org.apache.rocketmq.common.protocol.body.CMResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.CMResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;

@Data
 public class DlqMessageResendResult {
--- a/src/main/java/org/apache/rocketmq/dashboard/model/GroupConsumeInfo.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/GroupConsumeInfo.java
@@ -16,8 +16,8 @@
 */
 package org.apache.rocketmq.dashboard.model;

-import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
-import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
+import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

 public class GroupConsumeInfo implements Comparable<GroupConsumeInfo> {
    private String group;
--- a/src/main/java/org/apache/rocketmq/dashboard/model/QueueStatInfo.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/QueueStatInfo.java
@@ -16,7 +16,7 @@
 */
 package org.apache.rocketmq.dashboard.model;

-import org.apache.rocketmq.common.admin.OffsetWrapper;
+import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.springframework.beans.BeanUtils;

--- a/src/main/java/org/apache/rocketmq/dashboard/model/request/ConsumerConfigInfo.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/request/ConsumerConfigInfo.java
@@ -16,7 +16,7 @@
 */
 package org.apache.rocketmq.dashboard.model.request;

-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;

 import java.util.List;

--- a/src/main/java/org/apache/rocketmq/dashboard/model/request/TopicConfigInfo.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/model/request/TopicConfigInfo.java
@@ -15,7 +15,6 @@
 * limitations under the License.
 */
 package org.apache.rocketmq.dashboard.model.request;
-
 import com.google.common.base.Objects;

 import java.util.List;
@@ -32,6 +31,7 @@ public class TopicConfigInfo {
    private int perm;
    private boolean order;

+    private String messageType;
    public List<String> getClusterNameList() {
        return clusterNameList;
    }
@@ -92,6 +92,18 @@ public class TopicConfigInfo {
        this.order = order;
    }

+
+    public String getMessageType() {
+        return messageType;
+    }
+
+    public void setMessageType(String messageType) {
+        this.messageType = messageType;
+    }
+
+
+
+
    @Override
    public boolean equals(Object o) {
        if (this == o)
@@ -103,12 +115,13 @@ public class TopicConfigInfo {
            readQueueNums == that.readQueueNums &&
            perm == that.perm &&
            order == that.order &&
-            Objects.equal(topicName, that.topicName);
+            Objects.equal(topicName, that.topicName) &&
+            Objects.equal(messageType, that.messageType);
    }

    @Override
    public int hashCode() {
-        return Objects.hashCode(topicName, writeQueueNums, readQueueNums, perm, order);
+        return Objects.hashCode(topicName, writeQueueNums, readQueueNums, perm, order,messageType);
    }

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

 package org.apache.rocketmq.dashboard.service;

-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
 import org.apache.rocketmq.dashboard.model.ConsumerGroupRollBackStat;
 import org.apache.rocketmq.dashboard.model.GroupConsumeInfo;
 import org.apache.rocketmq.dashboard.model.TopicConsumerInfo;
--- a/src/main/java/org/apache/rocketmq/dashboard/service/MessageService.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/MessageService.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.service;

 import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 import org.apache.rocketmq.dashboard.model.MessagePage;
 import org.apache.rocketmq.dashboard.model.request.MessageQuery;
 import org.apache.rocketmq.tools.admin.api.MessageTrack;
--- a/src/main/java/org/apache/rocketmq/dashboard/service/ProducerService.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/ProducerService.java
@@ -17,7 +17,7 @@

 package org.apache.rocketmq.dashboard.service;

-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;

 public interface ProducerService {
    ProducerConnection getProducerConnection(String producerGroup, String topic);
--- a/src/main/java/org/apache/rocketmq/dashboard/service/TopicService.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/TopicService.java
@@ -19,10 +19,10 @@ package org.apache.rocketmq.dashboard.service;

 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;

--- a/src/main/java/org/apache/rocketmq/dashboard/service/client/MQAdminExtImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/client/MQAdminExtImpl.java
@@ -29,31 +29,41 @@ import org.apache.rocketmq.client.impl.MQAdminImpl;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.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.protocol.RequestCode;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.ClusterAclVersionInfo;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
-import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
-import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.common.message.MessageRequestMode;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
+import org.apache.rocketmq.remoting.protocol.body.QueryConsumeQueueResponseBody;
+import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
+import org.apache.rocketmq.remoting.protocol.subscription.GroupForbidden;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.util.JsonUtil;
 import org.apache.rocketmq.remoting.RemotingClient;
 import org.apache.rocketmq.remoting.exception.RemotingCommandException;
@@ -63,7 +73,9 @@ 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;
@@ -81,7 +93,7 @@ public class MQAdminExtImpl implements MQAdminExt {
    @Override
    public void updateBrokerConfig(String brokerAddr, Properties properties)
        throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
-        UnsupportedEncodingException, InterruptedException, MQBrokerException {
+        UnsupportedEncodingException, InterruptedException, MQBrokerException, MQClientException {
        MQAdminInstance.threadLocalMQAdminExt().updateBrokerConfig(brokerAddr, properties);
    }

@@ -128,7 +140,7 @@ public class MQAdminExtImpl implements MQAdminExt {
    }

    @Override
-    public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) {
+    public SubscriptionGroupConfig examineSubscriptionGroupConfig(String addr, String group) throws MQBrokerException {
        RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_SUBSCRIPTIONGROUP_CONFIG, null);
        RemotingCommand response = null;
@@ -136,7 +148,8 @@ public class MQAdminExtImpl implements MQAdminExt {
            response = remotingClient.invokeSync(addr, request, 3000);
        }
        catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
        assert response != null;
        switch (response.getCode()) {
@@ -145,12 +158,12 @@ public class MQAdminExtImpl implements MQAdminExt {
                return subscriptionGroupWrapper.getSubscriptionGroupTable().get(group);
            }
            default:
-                throw Throwables.propagate(new MQBrokerException(response.getCode(), response.getRemark()));
+                throw new MQBrokerException(response.getCode(), response.getRemark());
        }
    }

    @Override
-    public TopicConfig examineTopicConfig(String addr, String topic) {
+    public TopicConfig examineTopicConfig(String addr, String topic) throws MQBrokerException {
        RemotingClient remotingClient = MQAdminInstance.threadLocalRemotingClient();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
        RemotingCommand response = null;
@@ -158,7 +171,8 @@ public class MQAdminExtImpl implements MQAdminExt {
            response = remotingClient.invokeSync(addr, request, 3000);
        }
        catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
@@ -166,7 +180,7 @@ public class MQAdminExtImpl implements MQAdminExt {
                return topicConfigSerializeWrapper.getTopicConfigTable().get(topic);
            }
            default:
-                throw Throwables.propagate(new MQBrokerException(response.getCode(), response.getRemark()));
+                throw new MQBrokerException(response.getCode(), response.getRemark());
        }
    }

@@ -376,14 +390,14 @@ public class MQAdminExtImpl implements MQAdminExt {
    }

    @Override
-    public void createTopic(String key, String newTopic, int queueNum) throws MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum);
+    public void createTopic(String key, String newTopic, int queueNum, Map<String, String> attributes) throws MQClientException {
+        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, attributes);
    }

    @Override
-    public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag)
+    public void createTopic(String key, String newTopic, int queueNum, int topicSysFlag, Map<String, String> attributes)
        throws MQClientException {
-        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag);
+        MQAdminInstance.threadLocalMQAdminExt().createTopic(key, newTopic, queueNum, topicSysFlag, attributes);
    }

    @Override
@@ -572,4 +586,257 @@ public class MQAdminExtImpl implements MQAdminExt {
        String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
        return false;
    }
+
+    @Override
+    public void addBrokerToContainer(String brokerContainerAddr, String brokerConfig) throws InterruptedException,
+            MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'addBrokerToContainer'");
+    }
+
+    @Override
+    public void removeBrokerFromContainer(String brokerContainerAddr, String clusterName, String brokerName,
+            long brokerId) throws InterruptedException, MQBrokerException, RemotingTimeoutException,
+            RemotingSendRequestException, RemotingConnectException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'removeBrokerFromContainer'");
+    }
+
+    @Override
+    public void updateGlobalWhiteAddrConfig(String addr, String globalWhiteAddrs, String aclFileFullPath)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'updateGlobalWhiteAddrConfig'");
+    }
+
+    @Override
+    public TopicStatsTable examineTopicStats(String brokerAddr, String topic)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineTopicStats'");
+    }
+
+    @Override
+    public AdminToolResult<TopicStatsTable> examineTopicStatsConcurrent(String topic) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineTopicStatsConcurrent'");
+    }
+
+    @Override
+    public ConsumeStats examineConsumeStats(String brokerAddr, String consumerGroup, String topicName,
+            long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStats'");
+    }
+
+    @Override
+    public AdminToolResult<ConsumeStats> examineConsumeStatsConcurrent(String consumerGroup, String topic) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineConsumeStatsConcurrent'");
+    }
+
+    @Override
+    public ConsumerConnection examineConsumerConnectionInfo(String consumerGroup, String brokerAddr)
+            throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'examineConsumerConnectionInfo'");
+    }
+
+    @Override
+    public ProducerTableInfo getAllProducerInfo(String brokerAddr)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getAllProducerInfo'");
+    }
+
+    @Override
+    public void deleteTopic(String topicName, String clusterName)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteTopic'");
+    }
+
+    @Override
+    public AdminToolResult<BrokerOperatorResult> deleteTopicInBrokerConcurrent(Set<String> addrs, String topic) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteTopicInBrokerConcurrent'");
+    }
+
+    @Override
+    public void deleteTopicInNameServer(Set<String> addrs, String clusterName, String topic)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteTopicInNameServer'");
+    }
+
+    @Override
+    public AdminToolResult<BrokerOperatorResult> resetOffsetNewConcurrent(String group, String topic, long timestamp) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'resetOffsetNewConcurrent'");
+    }
+
+    @Override
+    public TopicList queryTopicsByConsumer(String group)
+            throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryTopicsByConsumer'");
+    }
+
+    @Override
+    public AdminToolResult<TopicList> queryTopicsByConsumerConcurrent(String group) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryTopicsByConsumerConcurrent'");
+    }
+
+    @Override
+    public SubscriptionData querySubscription(String group, String topic)
+            throws InterruptedException, MQBrokerException, RemotingException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'querySubscription'");
+    }
+
+    @Override
+    public AdminToolResult<List<QueueTimeSpan>> queryConsumeTimeSpanConcurrent(String topic, String group) {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryConsumeTimeSpanConcurrent'");
+    }
+
+    @Override
+    public boolean deleteExpiredCommitLog(String cluster) throws RemotingConnectException, RemotingSendRequestException,
+            RemotingTimeoutException, MQClientException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteExpiredCommitLog'");
+    }
+
+    @Override
+    public boolean deleteExpiredCommitLogByAddr(String addr) throws RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException, MQClientException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'deleteExpiredCommitLogByAddr'");
+    }
+
+    @Override
+    public ConsumerRunningInfo getConsumerRunningInfo(String consumerGroup, String clientId, boolean jstack,
+            boolean metrics) throws RemotingException, MQClientException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getConsumerRunningInfo'");
+    }
+
+    @Override
+    public List<MessageTrack> messageTrackDetailConcurrent(MessageExt msg)
+            throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'messageTrackDetailConcurrent'");
+    }
+
+    @Override
+    public void setMessageRequestMode(String brokerAddr, String topic, String consumerGroup, MessageRequestMode mode,
+            int popWorkGroupSize, long timeoutMillis) throws InterruptedException, RemotingTimeoutException,
+            RemotingSendRequestException, RemotingConnectException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'setMessageRequestMode'");
+    }
+
+    @Override
+    public long searchOffset(String brokerAddr, String topicName, int queueId, long timestamp, long timeoutMillis)
+            throws RemotingException, MQBrokerException, InterruptedException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'searchOffset'");
+    }
+
+    @Override
+    public void resetOffsetByQueueId(String brokerAddr, String consumerGroup, String topicName, int queueId,
+            long resetOffset) throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'resetOffsetByQueueId'");
+    }
+
+    @Override
+    public void createStaticTopic(String addr, String defaultTopic, TopicConfig topicConfig,
+            TopicQueueMappingDetail mappingDetail, boolean force)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'createStaticTopic'");
+    }
+
+    @Override
+    public GroupForbidden updateAndGetGroupReadForbidden(String brokerAddr, String groupName, String topicName,
+            Boolean readable) throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'updateAndGetGroupReadForbidden'");
+    }
+
+    @Override
+    public MessageExt queryMessage(String clusterName, String topic, String msgId)
+            throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'queryMessage'");
+    }
+
+    @Override
+    public HARuntimeInfo getBrokerHAStatus(String brokerAddr) throws RemotingConnectException,
+            RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getBrokerHAStatus'");
+    }
+
+    @Override
+    public BrokerReplicasInfo getInSyncStateData(String controllerAddress, List<String> brokers)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getInSyncStateData'");
+    }
+
+    @Override
+    public EpochEntryCache getBrokerEpochCache(String brokerAddr)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getBrokerEpochCache'");
+    }
+
+    @Override
+    public GetMetaDataResponseHeader getControllerMetaData(String controllerAddr)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getControllerMetaData'");
+    }
+
+    @Override
+    public void resetMasterFlushOffset(String brokerAddr, long masterFlushOffset) throws InterruptedException,
+            MQBrokerException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'resetMasterFlushOffset'");
+    }
+
+    @Override
+    public Map<String, Properties> getControllerConfig(List<String> controllerServers)
+            throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
+            RemotingConnectException, MQClientException, UnsupportedEncodingException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'getControllerConfig'");
+    }
+
+    @Override
+    public void updateControllerConfig(Properties properties, List<String> controllers)
+            throws InterruptedException, RemotingConnectException, UnsupportedEncodingException,
+            RemotingSendRequestException, RemotingTimeoutException, MQClientException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'updateControllerConfig'");
+    }
+
+    @Override
+    public ElectMasterResponseHeader electMaster(String controllerAddr, String clusterName, String brokerName,
+            String brokerAddr) throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'electMaster'");
+    }
+
+    @Override
+    public void cleanControllerBrokerData(String controllerAddr, String clusterName, String brokerName,
+            String brokerAddr, boolean isCleanLivingBroker)
+            throws RemotingException, InterruptedException, MQBrokerException {
+        // TODO Auto-generated method stub
+        throw new UnsupportedOperationException("Unimplemented method 'cleanControllerBrokerData'");
+    }
 }
--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/AclServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/AclServiceImpl.java
@@ -36,8 +36,8 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.dashboard.model.request.AclRequest;
 import org.apache.rocketmq.dashboard.service.AbstractCommonService;
 import org.apache.rocketmq.dashboard.service.AclService;
@@ -68,7 +68,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
            }
        } catch (Exception e) {
            log.error("getAclConfig error.", e);
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        AclConfig aclConfig = new AclConfig();
        aclConfig.setGlobalWhiteAddrs(Collections.emptyList());
@@ -100,7 +101,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }

    }
@@ -116,7 +118,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                log.info("Delete acl [{}] from broker [{}] complete", config.getAccessKey(), addr);
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -142,7 +145,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -174,7 +178,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                }
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -206,7 +211,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                }
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -249,7 +255,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                }
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }

    }
@@ -261,7 +268,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                mqAdminExt.createAndUpdatePlainAccessConfig(addr, config);
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -281,7 +289,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(aclConfig.getGlobalWhiteAddrs(), ","));
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -297,7 +306,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(aclConfig.getGlobalWhiteAddrs(), ","));
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -311,7 +321,8 @@ public class AclServiceImpl extends AbstractCommonService implements AclService
                mqAdminExt.updateGlobalWhiteAddrConfig(addr, StringUtils.join(whiteList, ","));
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/ClusterServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/ClusterServiceImpl.java
@@ -17,9 +17,10 @@

 package org.apache.rocketmq.dashboard.service.impl;

-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
 import org.apache.rocketmq.dashboard.service.ClusterService;
 import org.apache.rocketmq.dashboard.util.JsonUtil;
@@ -30,8 +31,10 @@ import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;

 import javax.annotation.Resource;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;

@Service
 public class ClusterServiceImpl implements ClusterService {
@@ -56,10 +59,14 @@ public class ClusterServiceImpl implements ClusterService {
            }
            resultMap.put("clusterInfo", clusterInfo);
            resultMap.put("brokerServer", brokerServer);
+            // add messageType
+            resultMap.put("messageTypes", Arrays.stream(TopicMessageType.values()).sorted()
+                    .collect(Collectors.toMap(TopicMessageType::getValue, messageType ->String.format("MESSAGE_TYPE_%s",messageType.getValue()))));
            return resultMap;
        }
        catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
    }

@@ -70,7 +77,8 @@ public class ClusterServiceImpl implements ClusterService {
            return mqAdminExt.getBrokerConfig(brokerAddr);
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }
 }
--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/ConsumerServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/ConsumerServiceImpl.java
@@ -44,18 +44,18 @@ import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.common.utils.ThreadUtils;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.model.ConsumerGroupRollBackStat;
@@ -73,8 +73,6 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.stereotype.Service;

-import static com.google.common.base.Throwables.propagate;
-
@Service
 public class ConsumerServiceImpl extends AbstractCommonService implements ConsumerService, InitializingBean, DisposableBean {
    private Logger logger = LoggerFactory.getLogger(ConsumerServiceImpl.class);
@@ -131,7 +129,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            }
        }
        catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
        List<GroupConsumeInfo> groupConsumeInfoList = Collections.synchronizedList(Lists.newArrayList());
        CountDownLatch countDownLatch = new CountDownLatch(consumerGroupSet.size());
@@ -218,7 +217,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            consumeStats = mqAdminExt.examineConsumeStats(groupName, topic);
        }
        catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        List<MessageQueue> mqList = Lists.newArrayList(Iterables.filter(consumeStats.getOffsetTable().keySet(), new Predicate<MessageQueue>() {
            @Override
@@ -278,7 +278,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            return group2ConsumerInfoMap;
        }
        catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -341,7 +342,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            }
        }
        catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        return consumerConfigInfoList;
    }
@@ -366,7 +368,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            }
        }
        catch (Exception e) {
-            throw propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        return true;
    }
@@ -393,7 +396,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            }
        }
        catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
        return true;
    }
@@ -408,7 +412,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            }
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        return brokerNameSet;

@@ -420,7 +425,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            return mqAdminExt.examineConsumerConnectionInfo(consumerGroup);
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -430,7 +436,8 @@ public class ConsumerServiceImpl extends AbstractCommonService implements Consum
            return mqAdminExt.getConsumerRunningInfo(consumerGroup, clientId, jstack);
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }
 }
--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/DashboardCollectServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/DashboardCollectServiceImpl.java
@@ -107,7 +107,8 @@ public class DashboardCollectServiceImpl implements DashboardCollectService {
            strings = Files.readLines(file, Charsets.UTF_8);
        }
        catch (IOException e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        StringBuffer sb = new StringBuffer();
        for (String string : strings) {
--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/DlqMessageServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/DlqMessageServiceImpl.java
@@ -25,8 +25,8 @@ import javax.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
 import org.apache.rocketmq.dashboard.model.DlqMessageResendResult;
 import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
 import org.apache.rocketmq.dashboard.model.MessagePage;
@@ -62,10 +62,12 @@ public class DlqMessageServiceImpl implements DlqMessageService {
                && e.getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) {
                return new MessagePage(new PageImpl<>(messageViews, page, 0), query.getTaskId());
            } else {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        return messageService.queryMessageByPage(query);
    }
--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/MessageServiceImpl.java
@@ -37,9 +37,9 @@ import org.apache.rocketmq.common.Pair;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.exception.ServiceException;
 import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
@@ -115,7 +115,8 @@ public class MessageServiceImpl implements MessageService {
            if (err instanceof MQClientException) {
                throw new ServiceException(-1, ((MQClientException) err).getErrorMessage());
            }
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
    }

@@ -185,7 +186,8 @@ public class MessageServiceImpl implements MessageService {
            });
            return messageViewList;
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        } finally {
            consumer.shutdown();
        }
@@ -209,7 +211,8 @@ public class MessageServiceImpl implements MessageService {
            try {
                return mqAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
            } catch (Exception e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
            }
        }

@@ -223,7 +226,8 @@ public class MessageServiceImpl implements MessageService {
                return mqAdminExt.consumeMessageDirectly(consumerGroup, connection.getClientId(), topic, msgId);
            }
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        throw new IllegalStateException("NO CONSUMER");

@@ -388,7 +392,8 @@ public class MessageServiceImpl implements MessageService {
            PageImpl<MessageView> page = new PageImpl<>(messageViews, query.page(), total);
            return new MessagePageTask(page, queueOffsetInfos);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        } finally {
            consumer.shutdown();
        }
@@ -455,7 +460,8 @@ public class MessageServiceImpl implements MessageService {
            }
            return new PageImpl<>(messageViews, query.page(), total);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        } finally {
            consumer.shutdown();
        }
--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/MonitorServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/MonitorServiceImpl.java
@@ -82,7 +82,8 @@ public class MonitorServiceImpl implements MonitorService {
            MixAll.string2File(dataStr, path);
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

--- a/src/main/java/org/apache/rocketmq/dashboard/service/impl/ProducerServiceImpl.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/service/impl/ProducerServiceImpl.java
@@ -19,7 +19,7 @@ package org.apache.rocketmq.dashboard.service.impl;

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

 import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
-import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.TransactionListener;
+import org.apache.rocketmq.client.producer.TransactionMQProducer;
+import org.apache.rocketmq.client.producer.LocalTransactionState;
 import org.apache.rocketmq.client.trace.TraceContext;
 import org.apache.rocketmq.client.trace.TraceDispatcher;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
 import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
@@ -44,6 +43,12 @@ 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.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.tools.command.CommandUtil;
 import org.joor.Reflect;
 import org.springframework.beans.BeanUtils;
@@ -55,6 +60,11 @@ import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import static org.apache.rocketmq.common.TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE;

@Service
 public class TopicServiceImpl extends AbstractCommonService implements TopicService {
@@ -68,23 +78,24 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
            TopicList allTopics = mqAdminExt.fetchAllTopicList();
            TopicList sysTopics = getSystemTopicList();
            Set<String> topics =
-                allTopics.getTopicList().stream().map(topic -> {
-                    if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
-                        topic = String.format("%s%s", "%SYS%", topic);
-                    }
-                    return topic;
-                }).filter(topic -> {
-                    if (skipRetryAndDlq) {
-                        return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
-                            || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
-                    }
-                    return true;
-                }).collect(Collectors.toSet());
+                    allTopics.getTopicList().stream().map(topic -> {
+                        if (!skipSysProcess && sysTopics.getTopicList().contains(topic)) {
+                            topic = String.format("%s%s", "%SYS%", topic);
+                        }
+                        return topic;
+                    }).filter(topic -> {
+                        if (skipRetryAndDlq) {
+                            return !(topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)
+                                    || topic.startsWith(MixAll.DLQ_GROUP_TOPIC_PREFIX));
+                        }
+                        return true;
+                    }).collect(Collectors.toSet());
            allTopics.getTopicList().clear();
            allTopics.getTopicList().addAll(topics);
            return allTopics;
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -93,7 +104,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
        try {
            return mqAdminExt.examineTopicStats(topic);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -102,7 +114,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
        try {
            return mqAdminExt.examineTopicRouteInfo(topic);
        } catch (Exception ex) {
-            throw Throwables.propagate(ex);
+            Throwables.throwIfUnchecked(ex);
+            throw new RuntimeException(ex);
        }
    }

@@ -111,7 +124,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
        try {
            return mqAdminExt.queryTopicConsumeByWho(topic);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -119,14 +133,20 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
    public void createOrUpdate(TopicConfigInfo topicCreateOrUpdateRequest) {
        TopicConfig topicConfig = new TopicConfig();
        BeanUtils.copyProperties(topicCreateOrUpdateRequest, topicConfig);
+        String messageType = topicCreateOrUpdateRequest.getMessageType();
+        if (StringUtils.isBlank(messageType)) {
+            messageType = TopicMessageType.NORMAL.name();
+        }
+        topicConfig.setAttributes(ImmutableMap.of("+".concat(TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName()), messageType));
        try {
            ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
            for (String brokerName : changeToBrokerNameSet(clusterInfo.getClusterAddrTable(),
-                topicCreateOrUpdateRequest.getClusterNameList(), topicCreateOrUpdateRequest.getBrokerNameList())) {
+                    topicCreateOrUpdateRequest.getClusterNameList(), topicCreateOrUpdateRequest.getBrokerNameList())) {
                mqAdminExt.createAndUpdateTopicConfig(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(), topicConfig);
            }
        } catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
    }

@@ -137,7 +157,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
            clusterInfo = mqAdminExt.examineBrokerClusterInfo();
            return mqAdminExt.examineTopicConfig(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr(), topic);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -150,6 +171,11 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
            TopicConfig topicConfig = examineTopicConfig(topic, brokerData.getBrokerName());
            BeanUtils.copyProperties(topicConfig, topicConfigInfo);
            topicConfigInfo.setBrokerNameList(Lists.newArrayList(brokerData.getBrokerName()));
+            String messageType = topicConfig.getAttributes().get(TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName());
+            if (StringUtils.isBlank(messageType)) {
+                messageType = TopicMessageType.UNSPECIFIED.name();
+            }
+            topicConfigInfo.setMessageType(messageType);
            topicConfigInfoList.add(topicConfigInfo);
        }
        return topicConfigInfoList;
@@ -170,7 +196,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
            }
            mqAdminExt.deleteTopicInNameServer(nameServerSet, topic);
        } catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
        return true;
    }
@@ -181,7 +208,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
        try {
            clusterInfo = mqAdminExt.examineBrokerClusterInfo();
        } catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
        for (String clusterName : clusterInfo.getClusterAddrTable().keySet()) {
            deleteTopic(topic, clusterName);
@@ -197,11 +225,13 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
            try {
                clusterInfo = mqAdminExt.examineBrokerClusterInfo();
            } catch (Exception e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
            }
            mqAdminExt.deleteTopicInBroker(Sets.newHashSet(clusterInfo.getBrokerAddrTable().get(brokerName).selectBrokerAddr()), topic);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
        return true;
    }
@@ -216,6 +246,12 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
        return defaultMQProducer;
    }

+    public TransactionMQProducer buildTransactionMQProducer(String producerGroup, RPCHook rpcHook, boolean traceEnabled) {
+        TransactionMQProducer defaultMQProducer = new TransactionMQProducer(null, producerGroup, rpcHook, traceEnabled, TopicValidator.RMQ_SYS_TRACE_TOPIC);
+        defaultMQProducer.setUseTLS(configure.isUseTLS());
+        return defaultMQProducer;
+    }
+
    private TopicList getSystemTopicList() {
        RPCHook rpcHook = null;
        boolean isEnableAcl = !StringUtils.isEmpty(configure.getAccessKey()) && !StringUtils.isEmpty(configure.getSecretKey());
@@ -230,7 +266,8 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ
            producer.start();
            return producer.getDefaultMQProducerImpl().getmQClientFactory().getMQClientAPIImpl().getSystemTopicList(20000L);
        } catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        } finally {
            producer.shutdown();
        }
@@ -238,31 +275,61 @@ public class TopicServiceImpl extends AbstractCommonService implements TopicServ

    @Override
    public SendResult sendTopicMessageRequest(SendTopicMessageRequest sendTopicMessageRequest) {
-        DefaultMQProducer producer = null;
+        List<TopicConfigInfo> topicConfigInfos = examineTopicConfig(sendTopicMessageRequest.getTopic());
+        String messageType = topicConfigInfos.get(0).getMessageType();
        AclClientRPCHook rpcHook = null;
        if (configure.isACLEnabled()) {
            rpcHook = new AclClientRPCHook(new SessionCredentials(
-                configure.getAccessKey(),
-                configure.getSecretKey()
+                    configure.getAccessKey(),
+                    configure.getSecretKey()
            ));
        }
-        producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
-        producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
-        producer.setNamesrvAddr(configure.getNamesrvAddr());
-        try {
-            producer.start();
-            Message msg = new Message(sendTopicMessageRequest.getTopic(),
-                sendTopicMessageRequest.getTag(),
-                sendTopicMessageRequest.getKey(),
-                sendTopicMessageRequest.getMessageBody().getBytes()
-            );
-            return producer.send(msg);
-        } catch (Exception e) {
-            throw Throwables.propagate(e);
-        } finally {
-            waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
-            producer.shutdown();
+        if (TopicMessageType.TRANSACTION.getValue().equals(messageType)) {
+            // transaction message
+            TransactionListener transactionListener = new TransactionListenerImpl();
+
+            TransactionMQProducer producer = buildTransactionMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
+            producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
+            producer.setNamesrvAddr(configure.getNamesrvAddr());
+            producer.setTransactionListener(transactionListener);
+            try {
+                producer.start();
+                Message msg = new Message(sendTopicMessageRequest.getTopic(),
+                        sendTopicMessageRequest.getTag(),
+                        sendTopicMessageRequest.getKey(),
+                        sendTopicMessageRequest.getMessageBody().getBytes()
+                );
+                return producer.sendMessageInTransaction(msg, null);
+            } catch (Exception e) {
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
+            } finally {
+                waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
+                producer.shutdown();
+            }
+        } else {
+            // no transaction message
+            DefaultMQProducer producer = null;
+            producer = buildDefaultMQProducer(MixAll.SELF_TEST_PRODUCER_GROUP, rpcHook, sendTopicMessageRequest.isTraceEnabled());
+            producer.setInstanceName(String.valueOf(System.currentTimeMillis()));
+            producer.setNamesrvAddr(configure.getNamesrvAddr());
+            try {
+                producer.start();
+                Message msg = new Message(sendTopicMessageRequest.getTopic(),
+                        sendTopicMessageRequest.getTag(),
+                        sendTopicMessageRequest.getKey(),
+                        sendTopicMessageRequest.getMessageBody().getBytes()
+                );
+                return producer.send(msg);
+            } catch (Exception e) {
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
+            } finally {
+                waitSendTraceFinish(producer, sendTopicMessageRequest.isTraceEnabled());
+                producer.shutdown();
+            }
        }
+
    }

    private void waitSendTraceFinish(DefaultMQProducer producer, boolean traceEnabled) {
@@ -284,4 +351,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;
+        }
+    }
 }
--- a/src/main/java/org/apache/rocketmq/dashboard/task/CollectTaskRunnble.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/task/CollectTaskRunnble.java
@@ -24,10 +24,10 @@ import java.util.List;
 import java.util.concurrent.ExecutionException;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.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;
@@ -93,7 +93,8 @@ public class CollectTaskRunnble implements Runnable {
            try {
                list = dashboardCollectService.getTopicMap().get(topic);
            } catch (ExecutionException e) {
-                throw Throwables.propagate(e);
+                Throwables.throwIfUnchecked(e);
+                throw new RuntimeException(e);
            }
            if (null == list) {
                list = Lists.newArrayList();
--- a/src/main/java/org/apache/rocketmq/dashboard/task/DashboardCollectTask.java
+++ b/src/main/java/org/apache/rocketmq/dashboard/task/DashboardCollectTask.java
@@ -34,10 +34,10 @@ import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import javax.annotation.Resource;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
 import org.apache.rocketmq.common.topic.TopicValidator;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.service.DashboardCollectService;
@@ -84,7 +84,8 @@ public class DashboardCollectTask {
            }
        }
        catch (Exception err) {
-            throw Throwables.propagate(err);
+            Throwables.throwIfUnchecked(err);
+            throw new RuntimeException(err);
        }
    }

@@ -128,7 +129,8 @@ public class DashboardCollectTask {
            log.debug("Broker Collected Data in memory = {}" + JsonUtil.obj2String(dashboardCollectService.getBrokerMap().asMap()));
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -144,10 +146,12 @@ public class DashboardCollectTask {
                Thread.sleep(1000);
            }
            catch (InterruptedException e1) {
-                throw Throwables.propagate(e1);
+                Throwables.throwIfUnchecked(e1);
+                throw new RuntimeException(e1);
            }
            fetchBrokerRuntimeStats(brokerAddr, retryTime - 1);
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -197,7 +201,8 @@ public class DashboardCollectTask {

        }
        catch (IOException e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -250,7 +255,7 @@ public class DashboardCollectTask {

    private void addSystemTopic() throws Exception {
        ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
-        HashMap<String, Set<String>> clusterTable = clusterInfo.getClusterAddrTable();
+        Map<String, Set<String>> clusterTable = clusterInfo.getClusterAddrTable();
        for (Map.Entry<String, Set<String>> entry : clusterTable.entrySet()) {
            String clusterName = entry.getKey();
            TopicValidator.addSystemTopic(clusterName);
--- a/src/main/resources/application.yml
+++ b/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
@@ -58,8 +60,8 @@ rocketmq:
    loginRequired: false
    useTLS: false
    # 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:
--- a/src/main/resources/static/src/i18n/en.js
+++ b/src/main/resources/static/src/i18n/en.js
@@ -123,5 +123,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",
 }
--- a/src/main/resources/static/src/i18n/zh.js
+++ b/src/main/resources/static/src/i18n/zh.js
@@ -124,5 +124,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": "事务消息",
 }
--- a/src/main/resources/static/src/topic.js
+++ b/src/main/resources/static/src/topic.js
@@ -328,8 +328,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 +355,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
                    }
--- a/src/main/resources/static/view/pages/cluster.html
+++ b/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>
--- a/src/main/resources/static/view/pages/topic.html
+++ b/src/main/resources/static/view/pages/topic.html
@@ -63,6 +63,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 +190,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">
--- a/src/test/java/org/apache/rocketmq/dashboard/admin/MQAdminExtImplTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/admin/MQAdminExtImplTest.java
@@ -34,27 +34,27 @@ import org.apache.rocketmq.client.impl.MQClientAPIImpl;
 import org.apache.rocketmq.client.impl.factory.MQClientInstance;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
-import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicConfigSerializeWrapper;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.service.client.MQAdminExtImpl;
 import org.apache.rocketmq.dashboard.service.client.MQAdminInstance;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
@@ -82,6 +82,7 @@ import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
@@ -570,11 +571,13 @@ public class MQAdminExtImplTest {
    public void testCreateTopic() throws Exception {
        assertNotNull(mqAdminExtImpl);
        {
-            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt());
-            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyInt());
+            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyMap());
+            doNothing().when(defaultMQAdminExt).createTopic(anyString(), anyString(), anyInt(), anyInt(), anyMap());
        }
-        mqAdminExtImpl.createTopic("key", "topic_test", 8);
-        mqAdminExtImpl.createTopic("key", "topic_test", 8, 1);
+        Map<String, String> map = new HashMap<>();
+        map.put("message.type", "FIFO");
+        mqAdminExtImpl.createTopic("key", "topic_test", 8, map);
+        mqAdminExtImpl.createTopic("key", "topic_test", 8, 1, map);
    }

    @Test
--- a/src/test/java/org/apache/rocketmq/dashboard/admin/MQAdminPoolTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/admin/MQAdminPoolTest.java
@@ -18,7 +18,7 @@ package org.apache.rocketmq.dashboard.admin;

 import org.apache.commons.pool2.PooledObject;
 import org.apache.commons.pool2.impl.GenericObjectPool;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
 import org.apache.rocketmq.tools.admin.MQAdminExt;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/AclControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/AclControllerTest.java
@@ -21,7 +21,7 @@ import com.google.common.collect.Lists;
 import java.util.List;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
 import org.apache.rocketmq.dashboard.model.request.AclRequest;
 import org.apache.rocketmq.dashboard.service.impl.AclServiceImpl;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/ClusterControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/ClusterControllerTest.java
@@ -18,8 +18,8 @@ package org.apache.rocketmq.dashboard.controller;

 import java.util.HashMap;
 import java.util.Properties;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
 import org.apache.rocketmq.dashboard.service.impl.ClusterServiceImpl;
 import org.apache.rocketmq.dashboard.util.MockObjectUtil;
 import org.junit.Test;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/ConsumerControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/ConsumerControllerTest.java
@@ -23,17 +23,17 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.RollbackStats;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.RollbackStats;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.SubscriptionGroupWrapper;
-import org.apache.rocketmq.common.protocol.heartbeat.ConsumeType;
-import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
+import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
 import org.apache.rocketmq.dashboard.model.request.DeleteSubGroupRequest;
 import org.apache.rocketmq.dashboard.model.request.ResetOffsetRequest;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/DlqMessageControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/DlqMessageControllerTest.java
@@ -21,10 +21,10 @@ import com.google.common.collect.Lists;
 import java.util.List;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.ResponseCode;
-import org.apache.rocketmq.common.protocol.body.CMResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.CMResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.model.DlqMessageRequest;
 import org.apache.rocketmq.dashboard.model.MessagePage;
 import org.apache.rocketmq.dashboard.model.MessageView;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/MessageControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/MessageControllerTest.java
@@ -32,9 +32,9 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.CMResult;
-import org.apache.rocketmq.common.protocol.body.ConsumeMessageDirectlyResult;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.CMResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
 import org.apache.rocketmq.dashboard.model.QueueOffsetInfo;
 import org.apache.rocketmq.dashboard.model.request.MessageQuery;
 import org.apache.rocketmq.dashboard.service.impl.MessageServiceImpl;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/ProducerControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/ProducerControllerTest.java
@@ -19,8 +19,8 @@ package org.apache.rocketmq.dashboard.controller;

 import java.util.HashSet;
 import org.apache.rocketmq.client.exception.MQClientException;
-import org.apache.rocketmq.common.protocol.body.Connection;
-import org.apache.rocketmq.common.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
 import org.apache.rocketmq.dashboard.interceptor.AuthInterceptor;
 import org.apache.rocketmq.dashboard.service.impl.LoginServiceImpl;
 import org.apache.rocketmq.dashboard.service.impl.ProducerServiceImpl;
--- a/src/test/java/org/apache/rocketmq/dashboard/controller/TopicControllerTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/controller/TopicControllerTest.java
@@ -29,16 +29,16 @@ import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.client.producer.SendStatus;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.admin.ConsumeStats;
-import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageQueue;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.ConsumerConnection;
-import org.apache.rocketmq.common.protocol.body.ConsumerRunningInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.model.request.SendTopicMessageRequest;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
 import org.apache.rocketmq.dashboard.service.impl.ConsumerServiceImpl;
--- a/src/test/java/org/apache/rocketmq/dashboard/task/DashboardCollectTaskTest.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/task/DashboardCollectTaskTest.java
@@ -33,12 +33,12 @@ import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.protocol.body.BrokerStatsData;
-import org.apache.rocketmq.common.protocol.body.ClusterInfo;
-import org.apache.rocketmq.common.protocol.body.GroupList;
-import org.apache.rocketmq.common.protocol.body.KVTable;
-import org.apache.rocketmq.common.protocol.body.TopicList;
-import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
 import org.apache.rocketmq.dashboard.BaseTest;
 import org.apache.rocketmq.dashboard.config.CollectExecutorConfig;
 import org.apache.rocketmq.dashboard.config.RMQConfigure;
--- a/src/test/java/org/apache/rocketmq/dashboard/testbase/RocketMQConsoleTestBase.java
+++ b/src/test/java/org/apache/rocketmq/dashboard/testbase/RocketMQConsoleTestBase.java
@@ -29,7 +29,7 @@ import org.apache.rocketmq.client.producer.SendResult;
 import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
 import org.apache.rocketmq.dashboard.model.request.ConsumerConfigInfo;
 import org.apache.rocketmq.dashboard.model.request.TopicConfigInfo;
 import org.apache.rocketmq.dashboard.service.ConsumerService;
@@ -78,7 +78,8 @@ public abstract class RocketMQConsoleTestBase {
                    }
                }
            }
-            throw Throwables.propagate(exception);
+            Throwables.throwIfUnchecked(exception);
+            throw new RuntimeException(exception);
        }
    }

@@ -91,7 +92,8 @@ public abstract class RocketMQConsoleTestBase {
            producer.start();
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

@@ -137,7 +139,8 @@ public abstract class RocketMQConsoleTestBase {
            consumer.start();
        }
        catch (Exception e) {
-            throw Throwables.propagate(e);
+            Throwables.throwIfUnchecked(e);
+            throw new RuntimeException(e);
        }
    }

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

-import static org.apache.rocketmq.common.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY;
+import static org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType.CONSUME_ACTIVELY;

 public class MockObjectUtil {
Author	SHA1	Message	Date
cndoit18	fc5e03cc50	Merge `a4a6000734` into `823bce2b8b`	2024-06-04 15:03:02 +08:00
guangdashao	823bce2b8b	feat: add topic message type add message type	2024-06-04 11:40:46 +08:00
Javen	2fb0fce0b1	perf: The new metrics of getTransferredTps for rocketmq5.x and the old metrics of getTransferedTps for rocketmq4.x (#197 ) Co-authored-by: jinwei2 <jinwei2@enmonster.com>	2024-03-26 17:02:16 +08:00
Abhijeet Mishra	6456630324	[#148 ] Throwables.propagate in deprecated for making runtime exception more verbose (#160 )	2023-04-19 20:33:17 +08:00
Abhijeet Mishra	a25ccd6337	5.1.0 rocketmq version update (#155 ) * update rocketmq version to 5.1.0	2023-04-07 08:30:18 +08:00
cndoit18	a4a6000734	fix(dockerfile): Unable to build docker image Fix the problem of not being able to build docker image Use multi-stage builds to optimize Signed-off-by: cndoit18 <cndoit18@outlook.com>	2023-01-06 19:31:57 +08:00