Finish the slave sync logic for topic queue mapping

broker/src/main/java/org/apache/rocketmq/broker/out/BrokerOuterAPI.java

@@ -350,7 +350,7 @@ public class BrokerOuterAPI {
         return changedList;
     }
 
-    public TopicConfigSerializeWrapper getAllTopicConfig(
+    public TopicConfigAndMappingSerializeWrapper getAllTopicConfig(
         final String addr) throws RemotingConnectException, RemotingSendRequestException,
         RemotingTimeoutException, InterruptedException, MQBrokerException {
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ALL_TOPIC_CONFIG, null);
@@ -359,7 +359,7 @@ public class BrokerOuterAPI {
         assert response != null;
         switch (response.getCode()) {
             case ResponseCode.SUCCESS: {
-                return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigSerializeWrapper.class);
+                return TopicConfigSerializeWrapper.decode(response.getBody(), TopicConfigAndMappingSerializeWrapper.class);
             }
             default:
                 break;

broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -64,6 +64,7 @@ import org.apache.rocketmq.common.protocol.body.QueryConsumeQueueResponseBody;
 import org.apache.rocketmq.common.protocol.body.QueryConsumeTimeSpanBody;
 import org.apache.rocketmq.common.protocol.body.QueryCorrectionOffsetBody;
 import org.apache.rocketmq.common.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.common.protocol.body.TopicConfigAndMappingSerializeWrapper;
 import org.apache.rocketmq.common.protocol.body.TopicList;
 import org.apache.rocketmq.common.protocol.body.UnlockBatchRequestBody;
 import org.apache.rocketmq.common.protocol.header.CloneGroupOffsetRequestHeader;
@@ -545,7 +546,16 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         // final GetAllTopicConfigResponseHeader responseHeader =
         // (GetAllTopicConfigResponseHeader) response.readCustomHeader();
 
-        String content = this.brokerController.getTopicConfigManager().encode();
+        TopicConfigAndMappingSerializeWrapper topicConfigAndMappingSerializeWrapper = new TopicConfigAndMappingSerializeWrapper();
+
+        topicConfigAndMappingSerializeWrapper.setDataVersion(this.brokerController.getTopicConfigManager().getDataVersion());
+        topicConfigAndMappingSerializeWrapper.setTopicConfigTable(this.brokerController.getTopicConfigManager().getTopicConfigTable());
+
+        topicConfigAndMappingSerializeWrapper.setMappingDataVersion(this.brokerController.getTopicQueueMappingManager().getDataVersion());
+        topicConfigAndMappingSerializeWrapper.setTopicQueueMappingDetailMap(this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable());
+
+
+        String content = topicConfigAndMappingSerializeWrapper.toJson();
         if (content != null && content.length() > 0) {
             try {
                 response.setBody(content.getBytes(MixAll.DEFAULT_CHARSET));

broker/src/main/java/org/apache/rocketmq/broker/slave/SlaveSynchronize.java

@@ -21,6 +21,7 @@ import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.common.protocol.body.TopicConfigAndMappingSerializeWrapper;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.common.protocol.body.ConsumerOffsetSerializeWrapper;
@@ -56,7 +57,7 @@ public class SlaveSynchronize {
         String masterAddrBak = this.masterAddr;
         if (masterAddrBak != null && !masterAddrBak.equals(brokerController.getBrokerAddr())) {
             try {
-                TopicConfigSerializeWrapper topicWrapper =
+                TopicConfigAndMappingSerializeWrapper topicWrapper =
                     this.brokerController.getBrokerOuterAPI().getAllTopicConfig(masterAddrBak);
                 if (!this.brokerController.getTopicConfigManager().getDataVersion()
                     .equals(topicWrapper.getDataVersion())) {
@@ -67,9 +68,17 @@ public class SlaveSynchronize {
                     this.brokerController.getTopicConfigManager().getTopicConfigTable()
                         .putAll(topicWrapper.getTopicConfigTable());
                     this.brokerController.getTopicConfigManager().persist();
-
-                    log.info("Update slave topic config from master, {}", masterAddrBak);
                 }
+                if (topicWrapper.getTopicQueueMappingDetailMap() != null
+                        && !topicWrapper.getMappingDataVersion().equals(this.brokerController.getTopicQueueMappingManager().getDataVersion())) {
+                    this.brokerController.getTopicQueueMappingManager().getDataVersion()
+                            .assignNewOne(topicWrapper.getMappingDataVersion());
+                    this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable().clear();
+                    this.brokerController.getTopicQueueMappingManager().getTopicQueueMappingTable()
+                            .putAll(topicWrapper.getTopicQueueMappingDetailMap());
+                    this.brokerController.getTopicQueueMappingManager().persist();
+                }
+                log.info("Update slave topic config from master, {}", masterAddrBak);
             } catch (Exception e) {
                 log.error("SyncTopicConfig Exception, {}", masterAddrBak, e);
             }

common/src/main/java/org/apache/rocketmq/common/protocol/body/TopicConfigAndMappingSerializeWrapper.java

@@ -17,6 +17,8 @@
 
 package org.apache.rocketmq.common.protocol.body;
 
+import org.apache.rocketmq.common.DataVersion;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 
 import java.util.Map;
@@ -25,6 +27,11 @@ import java.util.concurrent.ConcurrentHashMap;
 public class TopicConfigAndMappingSerializeWrapper extends TopicConfigSerializeWrapper {
     private Map<String/* topic */, TopicQueueMappingInfo> topicQueueMappingInfoMap = new ConcurrentHashMap<String, TopicQueueMappingInfo>();
 
+    private Map<String/* topic */, TopicQueueMappingDetail> topicQueueMappingDetailMap = new ConcurrentHashMap<String, TopicQueueMappingDetail>();
+
+    private DataVersion mappingDataVersion = new DataVersion();
+
+
     public Map<String, TopicQueueMappingInfo> getTopicQueueMappingInfoMap() {
         return topicQueueMappingInfoMap;
     }
@@ -33,6 +40,22 @@ public class TopicConfigAndMappingSerializeWrapper extends TopicConfigSerializeW
         this.topicQueueMappingInfoMap = topicQueueMappingInfoMap;
     }
 
+    public Map<String, TopicQueueMappingDetail> getTopicQueueMappingDetailMap() {
+        return topicQueueMappingDetailMap;
+    }
+
+    public void setTopicQueueMappingDetailMap(Map<String, TopicQueueMappingDetail> topicQueueMappingDetailMap) {
+        this.topicQueueMappingDetailMap = topicQueueMappingDetailMap;
+    }
+
+    public DataVersion getMappingDataVersion() {
+        return mappingDataVersion;
+    }
+
+    public void setMappingDataVersion(DataVersion mappingDataVersion) {
+        this.mappingDataVersion = mappingDataVersion;
+    }
+
     public static TopicConfigAndMappingSerializeWrapper from(TopicConfigSerializeWrapper wrapper) {
         if (wrapper instanceof  TopicConfigAndMappingSerializeWrapper) {
             return (TopicConfigAndMappingSerializeWrapper) wrapper;