broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java

@@ -95,7 +95,7 @@ import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.constant.PermName;

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

@@ -112,6 +112,10 @@ import org.apache.rocketmq.common.protocol.header.filtersrv.RegisterFilterServer
 import org.apache.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import org.apache.rocketmq.common.protocol.route.LogicalQueueRouteData;
 import org.apache.rocketmq.common.protocol.route.MessageQueueRouteState;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
+import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.stats.StatsItem;
 import org.apache.rocketmq.common.stats.StatsSnapshot;
 import org.apache.rocketmq.common.subscription.SubscriptionGroupConfig;

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

@@ -20,7 +20,7 @@ import io.netty.channel.ChannelHandlerContext;
 import java.util.List;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
-import org.apache.rocketmq.common.TopicQueueMappingContext;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.common.protocol.RequestCode;

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

@@ -36,12 +36,12 @@ import org.apache.rocketmq.broker.longpolling.PullRequest;
 import org.apache.rocketmq.broker.mqtrace.ConsumeMessageContext;
 import org.apache.rocketmq.broker.mqtrace.ConsumeMessageHook;
 import org.apache.rocketmq.broker.pagecache.ManyMessageTransfer;
-import org.apache.rocketmq.common.LogicQueueMappingItem;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.TopicFilterType;
-import org.apache.rocketmq.common.TopicQueueMappingContext;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.filter.ExpressionType;

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

@@ -28,13 +28,13 @@ import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.mqtrace.ConsumeMessageContext;
 import org.apache.rocketmq.broker.mqtrace.ConsumeMessageHook;
 import org.apache.rocketmq.broker.mqtrace.SendMessageContext;
-import org.apache.rocketmq.common.LogicQueueMappingItem;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.TopicFilterType;
-import org.apache.rocketmq.common.TopicQueueMappingContext;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.help.FAQUrl;

broker/src/main/java/org/apache/rocketmq/broker/topic/TopicQueueMappingManager.java

@@ -22,12 +22,12 @@ import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.BrokerPathConfigHelper;
 import org.apache.rocketmq.common.ConfigManager;
 import org.apache.rocketmq.common.DataVersion;
-import org.apache.rocketmq.common.LogicQueueMappingItem;
-import org.apache.rocketmq.common.TopicQueueMappingContext;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingContext;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.body.TopicQueueMappingSerializeWrapper;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 import org.apache.rocketmq.common.rpc.TopicQueueRequestHeader;

client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java

@@ -28,7 +28,7 @@ import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
 import org.apache.rocketmq.client.log.ClientLogger;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.help.FAQUrl;
 import org.apache.rocketmq.common.message.MessageClientIDSetter;
 import org.apache.rocketmq.common.message.MessageConst;

client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java

@@ -57,8 +57,8 @@ import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicConfigAndQueueMapping;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.admin.ConsumeStats;
 import org.apache.rocketmq.common.admin.TopicStatsTable;

client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java

@@ -61,7 +61,6 @@ import org.apache.rocketmq.client.stat.ConsumerStatsManager;
 import org.apache.rocketmq.common.MQVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceState;
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.filter.ExpressionType;

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

@@ -33,8 +33,7 @@ import java.util.zip.InflaterInputStream;
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;

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

@@ -17,7 +17,7 @@
 
 package org.apache.rocketmq.common.protocol.body;
 
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;

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

@@ -17,13 +17,13 @@
 
 package org.apache.rocketmq.common.protocol.body;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
 public class TopicConfigSerializeWrapper extends RemotingSerializable {
     private ConcurrentMap<String, TopicConfig> topicConfigTable =
         new ConcurrentHashMap<String, TopicConfig>();

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

@@ -16,7 +16,7 @@
  */
 package org.apache.rocketmq.common.protocol.body;
 
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
 public class TopicQueueMappingBody extends RemotingSerializable {

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

@@ -18,7 +18,7 @@
 package org.apache.rocketmq.common.protocol.body;
 
 import org.apache.rocketmq.common.DataVersion;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
 import java.util.Map;

common/src/main/java/org/apache/rocketmq/common/protocol/route/TopicRouteData.java

@@ -25,7 +25,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 
 public class TopicRouteData extends RemotingSerializable {

common/src/main/java/org/apache/rocketmq/common/rpc/ClientMetadata.java

 package org.apache.rocketmq.common.rpc;
 
 import org.apache.rocketmq.common.MixAll;
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.message.MessageQueue;
 import org.apache.rocketmq.common.protocol.body.ClusterInfo;

common/src/main/java/org/apache/rocketmq/common/LogicQueueMappingItem.java → common/src/main/java/org/apache/rocketmq/common/statictopic/LogicQueueMappingItem.java

-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
 
 public class LogicQueueMappingItem {
 
@@ -88,4 +88,22 @@ public class LogicQueueMappingItem {
     public long getTimeOfEnd() {
         return timeOfEnd;
     }
+
+    public void setLogicOffset(long logicOffset) {
+        this.logicOffset = logicOffset;
+    }
+
+    @Override
+    public String toString() {
+        return "LogicQueueMappingItem{" +
+                "gen=" + gen +
+                ", queueId=" + queueId +
+                ", bname='" + bname + '\'' +
+                ", logicOffset=" + logicOffset +
+                ", startOffset=" + startOffset +
+                ", endOffset=" + endOffset +
+                ", timeOfStart=" + timeOfStart +
+                ", timeOfEnd=" + timeOfEnd +
+                '}';
+    }
 }

common/src/main/java/org/apache/rocketmq/common/TopicConfigAndQueueMapping.java → common/src/main/java/org/apache/rocketmq/common/statictopic/TopicConfigAndQueueMapping.java

@@ -14,7 +14,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
+
+import org.apache.rocketmq.common.TopicConfig;
 
 public class TopicConfigAndQueueMapping extends TopicConfig {
     private TopicQueueMappingDetail mappingDetail;

common/src/main/java/org/apache/rocketmq/common/TopicQueueMappingContext.java → common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingContext.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
 
 import com.google.common.collect.ImmutableList;
 

common/src/main/java/org/apache/rocketmq/common/TopicQueueMappingDetail.java → common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingDetail.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
 
 import com.google.common.collect.ImmutableList;
 

common/src/main/java/org/apache/rocketmq/common/TopicQueueMappingInfo.java → common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingInfo.java

@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
 
 import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
 

common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingOne.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.common.statictopic;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+public class TopicQueueMappingOne extends RemotingSerializable {
+
+    String topic; // redundant field
+    String bname;  //identify the hosted broker name
+    Integer globalId;
+    ImmutableList<LogicQueueMappingItem> items;
+
+    public TopicQueueMappingOne(String topic, String bname, Integer globalId, ImmutableList<LogicQueueMappingItem> items) {
+        this.topic = topic;
+        this.bname = bname;
+        this.globalId = globalId;
+        this.items = items;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public String getBname() {
+        return bname;
+    }
+
+    public Integer getGlobalId() {
+        return globalId;
+    }
+
+    public ImmutableList<LogicQueueMappingItem> getItems() {
+        return items;
+    }
+}

common/src/main/java/org/apache/rocketmq/common/TopicQueueMappingUtils.java → common/src/main/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingUtils.java

@@ -14,12 +14,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
 
 import com.google.common.collect.ImmutableList;
+import org.apache.rocketmq.common.MixAll;
 
+import java.io.File;
 import java.util.AbstractMap;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -103,7 +106,63 @@ public class TopicQueueMappingUtils {
         return new AbstractMap.SimpleImmutableEntry<Long, Integer>(epoch, queueNum);
     }
 
-    public static Map<Integer, ImmutableList<LogicQueueMappingItem>> buildMappingItems(List<TopicQueueMappingDetail> mappingDetailList, boolean replace) {
+    public static List<TopicQueueMappingDetail> getMappingDetailFromConfig(Collection<TopicConfigAndQueueMapping> configs) {
+        List<TopicQueueMappingDetail> detailList = new ArrayList<TopicQueueMappingDetail>();
+        for (TopicConfigAndQueueMapping configMapping : configs) {
+            if (configMapping.getMappingDetail() != null) {
+                detailList.add(configMapping.getMappingDetail());
+            }
+        }
+        return detailList;
+    }
+
+    public static Map.Entry<Long, Integer> checkConsistenceOfTopicConfigAndQueueMapping(String topic, Map<String, TopicConfigAndQueueMapping> brokerConfigMap) {
+        if (brokerConfigMap == null
+            || brokerConfigMap.isEmpty()) {
+            return null;
+        }
+        //make sure it it not null
+        long maxEpoch = -1;
+        int maxNum = -1;
+        for (Map.Entry<String, TopicConfigAndQueueMapping> entry : brokerConfigMap.entrySet()) {
+            String broker = entry.getKey();
+            TopicConfigAndQueueMapping configMapping = entry.getValue();
+            if (configMapping.getMappingDetail() == null) {
+                throw new RuntimeException("Mapping info should not be null in broker " + broker);
+            }
+            TopicQueueMappingDetail mappingDetail = configMapping.getMappingDetail();
+            if (!broker.equals(mappingDetail.getBname())) {
+                throw new RuntimeException(String.format("The broker name is not equal %s != %s ", broker, mappingDetail.getBname()));
+            }
+            if (mappingDetail.isDirty()) {
+                throw new RuntimeException("The mapping info is dirty in broker  " + broker);
+            }
+            if (!configMapping.getTopicName().equals(mappingDetail.getTopic())) {
+                throw new RuntimeException("The topic name is inconsistent in broker  " + broker);
+            }
+            if (topic != null
+                && !topic.equals(mappingDetail.getTopic())) {
+                throw new RuntimeException("The topic name is not match for broker  " + broker);
+            }
+
+            if (maxEpoch != -1
+                && maxEpoch != mappingDetail.getEpoch()) {
+                throw new RuntimeException(String.format("epoch dose not match %d != %d in %s", maxEpoch, mappingDetail.getEpoch(), mappingDetail.getBname()));
+            } else {
+                maxEpoch = mappingDetail.getEpoch();
+            }
+
+            if (maxNum != -1
+                && maxNum != mappingDetail.getTotalQueues()) {
+                throw new RuntimeException(String.format("total queue number dose not match %d != %d in %s", maxNum, mappingDetail.getTotalQueues(), mappingDetail.getBname()));
+            } else {
+                maxNum = mappingDetail.getTotalQueues();
+            }
+        }
+        return new AbstractMap.SimpleEntry<Long, Integer>(maxEpoch, maxNum);
+    }
+
+    public static Map<Integer, TopicQueueMappingOne> checkAndBuildMappingItems(List<TopicQueueMappingDetail> mappingDetailList, boolean replace, boolean checkConsistence) {
         Collections.sort(mappingDetailList, new Comparator<TopicQueueMappingDetail>() {
             @Override
             public int compare(TopicQueueMappingDetail o1, TopicQueueMappingDetail o2) {
@@ -111,8 +170,12 @@ public class TopicQueueMappingUtils {
             }
         });
 
-        Map<Integer, ImmutableList<LogicQueueMappingItem>> globalIdMap = new HashMap<Integer, ImmutableList<LogicQueueMappingItem>>();
+        int maxNum = 0;
+        Map<Integer, TopicQueueMappingOne> globalIdMap = new HashMap<Integer, TopicQueueMappingOne>();
         for (TopicQueueMappingDetail mappingDetail : mappingDetailList) {
+            if (mappingDetail.totalQueues > maxNum) {
+                maxNum = mappingDetail.totalQueues;
+            }
             for (Map.Entry<Integer, ImmutableList<LogicQueueMappingItem>>  entry : mappingDetail.getHostedQueues().entrySet()) {
                 Integer globalid = entry.getKey();
                 String leaderBrokerName  = getLeaderBroker(entry.getValue());
@@ -125,7 +188,17 @@ public class TopicQueueMappingUtils {
                         throw new RuntimeException(String.format("The queue id is duplicated in broker %s %s", leaderBrokerName, mappingDetail.getBname()));
                     }
                 } else {
-                    globalIdMap.put(globalid, entry.getValue());
+                    globalIdMap.put(globalid, new TopicQueueMappingOne(mappingDetail.topic, mappingDetail.bname, globalid, entry.getValue()));
+                }
+            }
+        }
+        if (checkConsistence) {
+            if (maxNum != globalIdMap.size()) {
+                throw new RuntimeException(String.format("The total queue number in config dose not match the real hosted queues %d != %d", maxNum, globalIdMap.size()));
+            }
+            for (int i = 0; i < maxNum; i++) {
+                if (!globalIdMap.containsKey(i)) {
+                    throw new RuntimeException(String.format("The queue number %s is not in globalIdMap", i));
                 }
             }
         }
@@ -139,4 +212,23 @@ public class TopicQueueMappingUtils {
         assert items.size() > 0;
         return items.get(items.size() - 1);
     }
+
+    public static String writeToTemp(TopicRemappingDetailWrapper wrapper, boolean after) {
+        String topic = wrapper.getTopic();
+        String data = wrapper.toJson();
+        String suffix = TopicRemappingDetailWrapper.SUFFIX_BEFORE;
+        if (after) {
+            suffix = TopicRemappingDetailWrapper.SUFFIX_AFTER;
+        }
+        String fileName = System.getProperty("java.io.tmpdir") + File.separator + topic + "-" + wrapper.getEpoch() + suffix;
+        try {
+            MixAll.string2File(data, fileName);
+            return fileName;
+        } catch (Exception e) {
+            throw new RuntimeException("write file failed " + fileName,e);
+        }
+    }
+
+
+
 }

common/src/main/java/org/apache/rocketmq/common/statictopic/TopicRemappingDetailWrapper.java

+package org.apache.rocketmq.common.statictopic;
+
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+public class TopicRemappingDetailWrapper extends RemotingSerializable {
+    public static final String TYPE_CREATE_OR_UPDATE = "CREATE_OR_UPDATE";
+    public static final String TYPE_REMAPPING = "REMAPPING";
+
+    public static final String SUFFIX_BEFORE = ".before";
+    public static final String SUFFIX_AFTER = ".after";
+
+
+    private final String topic;
+    private final String type;
+    private final long epoch;
+    private Map<Integer, String> expectedIdToBroker = new HashMap<Integer, String>();
+
+    private Map<String, TopicConfigAndQueueMapping> brokerConfigMap = new HashMap<String, TopicConfigAndQueueMapping>();
+
+    private Set<String> brokerToMapIn = new HashSet<String>();
+
+    private Set<String> brokerToMapOut = new HashSet<String>();
+
+    public TopicRemappingDetailWrapper(String topic, String type, long epoch, Map<Integer, String> expectedIdToBroker, Map<String, TopicConfigAndQueueMapping> brokerConfigMap) {
+        this.topic = topic;
+        this.type = type;
+        this.epoch = epoch;
+        this.expectedIdToBroker = expectedIdToBroker;
+        this.brokerConfigMap = brokerConfigMap;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public long getEpoch() {
+        return epoch;
+    }
+
+    public Map<Integer, String> getExpectedIdToBroker() {
+        return expectedIdToBroker;
+    }
+
+    public Map<String, TopicConfigAndQueueMapping> getBrokerConfigMap() {
+        return brokerConfigMap;
+    }
+
+    public Set<String> getBrokerToMapIn() {
+        return brokerToMapIn;
+    }
+
+    public void setBrokerToMapIn(Set<String> brokerToMapIn) {
+        this.brokerToMapIn = brokerToMapIn;
+    }
+
+    public Set<String> getBrokerToMapOut() {
+        return brokerToMapOut;
+    }
+
+    public void setBrokerToMapOut(Set<String> brokerToMapOut) {
+        this.brokerToMapOut = brokerToMapOut;
+    }
+}

common/src/test/java/org/apache/rocketmq/common/TopicQueueMappingTest.java → common/src/test/java/org/apache/rocketmq/common/statictopic/TopicQueueMappingTest.java

-package org.apache.rocketmq.common;
+package org.apache.rocketmq.common.statictopic;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import com.google.common.collect.ImmutableList;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.io.File;
 import java.util.Map;
 
 public class TopicQueueMappingTest {
 
+    @Test
+    public void testWriteToFile() {
+        System.out.println(System.getProperty("java.io.tmpdir"));
+        System.out.println(File.separator);
+    }
+
+
     @Test
     public void testJsonSerialize() {
         LogicQueueMappingItem mappingItem = new LogicQueueMappingItem(1, 2, "broker01", 33333333333333333L, 44444444444444444L, 555555555555555555L, 6666666666666666L, 77777777777777777L);
@@ -28,7 +38,7 @@ public class TopicQueueMappingTest {
         Assert.assertEquals(mappingItemMap.get("timeOfStart"), mappingItem.getTimeOfStart());
         Assert.assertEquals(mappingItemMap.get("timeOfEnd"), mappingItem.getTimeOfEnd());
 
-        TopicQueueMappingDetail mappingDetail = new TopicQueueMappingDetail("test", 1, "broker01");
+        TopicQueueMappingDetail mappingDetail = new TopicQueueMappingDetail("test", 1, "broker01", System.currentTimeMillis());
         mappingDetail.putMappingInfo(0, ImmutableList.of(mappingItem));
 
         String mappingDetailJson = JSON.toJSONString(mappingDetail);

namesrv/src/main/java/org/apache/rocketmq/namesrv/routeinfo/RouteInfoManager.java

@@ -31,7 +31,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingInfo;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingInfo;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.protocol.RequestCode;

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java

@@ -28,7 +28,7 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.admin.ConsumeStats;
 import org.apache.rocketmq.common.admin.RollbackStats;
 import org.apache.rocketmq.common.admin.TopicStatsTable;
@@ -227,6 +227,12 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
         return defaultMQAdminExtImpl.examineTopicStats(topic);
     }
 
+    @Override
+    public TopicStatsTable examineTopicStats(String brokerAddr, String topic) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        return defaultMQAdminExtImpl.examineTopicStats(brokerAddr, topic);
+    }
+
     @Override
     public TopicList fetchAllTopicList() throws RemotingException, MQClientException, InterruptedException {
         return this.defaultMQAdminExtImpl.fetchAllTopicList();

tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java

@@ -42,7 +42,7 @@ import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.ServiceState;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.admin.ConsumeStats;
 import org.apache.rocketmq.common.admin.OffsetWrapper;
@@ -274,6 +274,13 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
         return topicStatsTable;
     }
 
+
+    @Override
+    public TopicStatsTable examineTopicStats(String brokerAddr, String topic) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException {
+        return this.mqClientInstance.getMQClientAPIImpl().getTopicStatsInfo(brokerAddr, topic, timeoutMillis);
+    }
+
     @Override
     public TopicList fetchAllTopicList() throws RemotingException, MQClientException, InterruptedException {
         return this.mqClientInstance.getMQClientAPIImpl().getTopicListFromNameServer(timeoutMillis);
@@ -1092,7 +1099,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
     public void createStaticTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig, final TopicQueueMappingDetail mappingDetail) throws MQClientException {
         this.mqClientInstance.getMQAdminImpl().createStaticTopic(addr, defaultTopic, topicConfig, mappingDetail);
     }
-    
+
     @Override
     public long searchOffset(MessageQueue mq, long timestamp) throws MQClientException {
         return this.mqClientInstance.getMQAdminImpl().searchOffset(mq, timestamp);

tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java

@@ -27,7 +27,7 @@ import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.AclConfig;
 import org.apache.rocketmq.common.PlainAccessConfig;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
 import org.apache.rocketmq.common.admin.ConsumeStats;
 import org.apache.rocketmq.common.admin.RollbackStats;
 import org.apache.rocketmq.common.admin.TopicStatsTable;
@@ -108,6 +108,9 @@ public interface MQAdminExt extends MQAdmin {
         final String topic) throws RemotingException, MQClientException, InterruptedException,
         MQBrokerException;
 
+    TopicStatsTable examineTopicStats(String brokerAddr, final String topic) throws RemotingException, MQClientException, InterruptedException,
+            MQBrokerException;
+
     TopicList fetchAllTopicList() throws RemotingException, MQClientException, InterruptedException;
 
     TopicList fetchTopicsByCLuster(

tools/src/main/java/org/apache/rocketmq/tools/command/logicalqueue/MigrateTopicLogicalQueueCommand.java

@@ -87,7 +87,7 @@ public class MigrateTopicLogicalQueueCommand implements SubCommand {
         String toBrokerName,
         Long forceDelta) throws RemotingException, MQBrokerException, InterruptedException, SubCommandException, MQClientException {
         TopicRouteData topicRouteInfo = mqAdminExt.examineTopicRouteInfo(topic);
-        LogicalQueuesInfo logicalQueuesInfo = topicRouteInfo.getLogicalQueuesInfo();
+        LogicalQueuesInfo logicalQueuesInfo =  null; /*topicRouteInfo.getLogicalQueuesInfo();*/
         if (logicalQueuesInfo == null) {
             throw new SubCommandException("topic not enabled logical queue");
         }

tools/src/main/java/org/apache/rocketmq/tools/command/topic/RemappingStaticTopicSubCommand.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.tools.command.topic;
+
+import com.google.common.collect.ImmutableList;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionGroup;
+import org.apache.commons.cli.Options;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
+import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingOne;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
+import org.apache.rocketmq.common.admin.TopicOffset;
+import org.apache.rocketmq.common.admin.TopicStatsTable;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.route.QueueData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
+import org.apache.rocketmq.common.rpc.ClientMetadata;
+import org.apache.rocketmq.common.statictopic.TopicRemappingDetailWrapper;
+import org.apache.rocketmq.remoting.RPCHook;
+import org.apache.rocketmq.srvutil.ServerUtil;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.apache.rocketmq.tools.command.SubCommand;
+import org.apache.rocketmq.tools.command.SubCommandException;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class RemappingStaticTopicSubCommand implements SubCommand {
+
+    @Override
+    public String commandName() {
+        return "updateStaticTopic";
+    }
+
+    @Override
+    public String commandDesc() {
+        return "Update or create static topic, which has fixed number of queues";
+    }
+
+    @Override
+    public Options buildCommandlineOptions(Options options) {
+        OptionGroup optionGroup = new OptionGroup();
+
+        Option opt = null;
+
+        opt = new Option("c", "clusters", true, "remapping static topic to clusters, comma separated");
+        optionGroup.addOption(opt);
+
+        opt = new Option("b", "brokers", true, "remapping static topic to brokers, comma separated");
+        optionGroup.addOption(opt);
+
+        optionGroup.setRequired(true);
+        options.addOptionGroup(optionGroup);
+
+        opt = new Option("t", "topic", true, "topic name");
+        opt.setRequired(true);
+        options.addOption(opt);
+
+        opt = new Option("f", "mapFile", true, "The map file name");
+        opt.setRequired(false);
+        options.addOption(opt);
+        return options;
+    }
+
+
+    public void executeFromFile(final CommandLine commandLine, final Options options,
+                                RPCHook rpcHook) throws SubCommandException {
+
+        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
+        defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
+        ClientMetadata clientMetadata = new ClientMetadata();
+
+        try {
+            String topic = commandLine.getOptionValue('t').trim();
+
+            String mapFileName = commandLine.getOptionValue('f').trim();
+            String mapData = MixAll.file2String(mapFileName);
+            TopicRemappingDetailWrapper wrapper = TopicRemappingDetailWrapper.decode(mapData.getBytes(), TopicRemappingDetailWrapper.class);
+            //double check the config
+            TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, wrapper.getBrokerConfigMap());
+            TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(wrapper.getBrokerConfigMap().values())), false, true);
+
+            ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
+            if (clusterInfo == null
+                    || clusterInfo.getClusterAddrTable().isEmpty()) {
+                throw new RuntimeException("The Cluster info is empty");
+            }
+            clientMetadata.refreshClusterInfo(clusterInfo);
+            doRemapping(topic, wrapper.getBrokerToMapIn(), wrapper.getBrokerToMapOut(), wrapper.getBrokerConfigMap(), clientMetadata, defaultMQAdminExt);
+            return;
+        }catch (Exception e) {
+            throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
+        } finally {
+            defaultMQAdminExt.shutdown();
+        }
+    }
+
+
+
+    public void doRemapping(String topic, Set<String> brokersToMapIn, Set<String> brokersToMapOut, Map<String, TopicConfigAndQueueMapping> brokerConfigMap,
+                            ClientMetadata clientMetadata, DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+        // now do the remapping
+        //Step1: let the new leader can be write without the logicOffset
+        for (String broker: brokersToMapIn) {
+            String addr = clientMetadata.findMasterBrokerAddr(broker);
+            TopicConfigAndQueueMapping configMapping = brokerConfigMap.get(broker);
+            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail());
+        }
+        //Step2: forbid the write of old leader
+        for (String broker: brokersToMapOut) {
+            String addr = clientMetadata.findMasterBrokerAddr(broker);
+            TopicConfigAndQueueMapping configMapping = brokerConfigMap.get(broker);
+            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail());
+        }
+        //Step3: decide the logic offset
+        for (String broker: brokersToMapOut) {
+            String addr = clientMetadata.findMasterBrokerAddr(broker);
+            TopicStatsTable statsTable = defaultMQAdminExt.examineTopicStats(addr, topic);
+            TopicConfigAndQueueMapping mapOutConfig = brokerConfigMap.get(broker);
+            for (Map.Entry<Integer, ImmutableList<LogicQueueMappingItem>> entry : mapOutConfig.getMappingDetail().getHostedQueues().entrySet()) {
+                ImmutableList<LogicQueueMappingItem> items = entry.getValue();
+                Integer globalId = entry.getKey();
+                if (items.size() < 2) {
+                    continue;
+                }
+                LogicQueueMappingItem newLeader = items.get(items.size() - 1);
+                LogicQueueMappingItem oldLeader = items.get(items.size() - 2);
+                if (newLeader.getLogicOffset()  > 0) {
+                    continue;
+                }
+                TopicOffset topicOffset = statsTable.getOffsetTable().get(new MessageQueue(topic, oldLeader.getBname(), oldLeader.getQueueId()));
+                if (topicOffset == null) {
+                    throw new RuntimeException("Cannot get the max offset for old leader " + oldLeader);
+                }
+                //TODO check the max offset, will it return -1?
+                if (topicOffset.getMaxOffset() < oldLeader.getStartOffset()) {
+                    throw new RuntimeException("The max offset is smaller then the start offset " + oldLeader + " " + topicOffset.getMaxOffset());
+                }
+                newLeader.setLogicOffset(oldLeader.computeStaticQueueOffset(topicOffset.getMaxOffset() + 10000));
+                TopicConfigAndQueueMapping mapInConfig = brokerConfigMap.get(newLeader.getBname());
+                //fresh the new leader
+                mapInConfig.getMappingDetail().putMappingInfo(globalId, items);
+            }
+        }
+        //Step4: write to the new leader with logic offset
+        for (String broker: brokersToMapIn) {
+            String addr = clientMetadata.findMasterBrokerAddr(broker);
+            TopicConfigAndQueueMapping configMapping = brokerConfigMap.get(broker);
+            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail());
+        }
+    }
+
+
+
+
+
+    @Override
+    public void execute(final CommandLine commandLine, final Options options,
+                        RPCHook rpcHook) throws SubCommandException {
+
+        if (!commandLine.hasOption('t')) {
+            ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
+            return;
+        }
+
+        if (commandLine.hasOption("f")) {
+            executeFromFile(commandLine, options, rpcHook);
+            return;
+        }
+
+        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
+        defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
+
+        ClientMetadata clientMetadata = new ClientMetadata();
+        Map<String, TopicConfigAndQueueMapping> brokerConfigMap = new HashMap<>();
+        Map<Integer, TopicQueueMappingOne> globalIdMap = new HashMap<>();
+        Set<String> targetBrokers = new HashSet<>();
+
+        try {
+            if ((!commandLine.hasOption("b") && !commandLine.hasOption('c'))) {
+                ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
+                return;
+            }
+            String topic = commandLine.getOptionValue('t').trim();
+
+            ClusterInfo clusterInfo  = defaultMQAdminExt.examineBrokerClusterInfo();
+            if (clusterInfo == null
+                    || clusterInfo.getClusterAddrTable().isEmpty()) {
+                throw new RuntimeException("The Cluster info is empty");
+            }
+            clientMetadata.refreshClusterInfo(clusterInfo);
+            {
+                if (commandLine.hasOption("b")) {
+                    String brokerStrs = commandLine.getOptionValue("b").trim();
+                    for (String broker: brokerStrs.split(",")) {
+                        targetBrokers.add(broker.trim());
+                    }
+                } else if (commandLine.hasOption("c")) {
+                    String clusters = commandLine.getOptionValue('c').trim();
+                    for (String cluster : clusters.split(",")) {
+                        cluster = cluster.trim();
+                        if (clusterInfo.getClusterAddrTable().get(cluster) != null) {
+                            targetBrokers.addAll(clusterInfo.getClusterAddrTable().get(cluster));
+                        }
+                    }
+                }
+                if (targetBrokers.isEmpty()) {
+                    throw new RuntimeException("Find none brokers, do nothing");
+                }
+                for (String broker : targetBrokers) {
+                    String addr = clientMetadata.findMasterBrokerAddr(broker);
+                    if (addr == null) {
+                        throw new RuntimeException("Can't find addr for broker " + broker);
+                    }
+                }
+            }
+
+            //get the existed topic config and mapping
+            {
+                TopicRouteData routeData = defaultMQAdminExt.examineTopicRouteInfo(topic);
+                clientMetadata.freshTopicRoute(topic, routeData);
+
+                if (routeData != null
+                        && !routeData.getQueueDatas().isEmpty()) {
+                    for (QueueData queueData: routeData.getQueueDatas()) {
+                        String bname = queueData.getBrokerName();
+                        String addr = clientMetadata.findMasterBrokerAddr(bname);
+                        TopicConfigAndQueueMapping mapping = (TopicConfigAndQueueMapping) defaultMQAdminExt.examineTopicConfig(addr, topic);
+                        //allow the config is null
+                        if (mapping != null) {
+                            brokerConfigMap.put(bname, mapping);
+                        }
+                    }
+                }
+            }
+
+            if (brokerConfigMap.isEmpty()) {
+                throw new RuntimeException("No topic route to do the remapping");
+            }
+
+            final Map.Entry<Long, Integer> maxEpochAndNum = TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, brokerConfigMap);
+            globalIdMap = TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
+            //the check is ok, now do the mapping allocation
+            int maxNum = maxEpochAndNum.getValue();
+            long maxEpoch = maxEpochAndNum.getKey();
+
+            {
+                TopicRemappingDetailWrapper oldWrapper = new TopicRemappingDetailWrapper(topic, TopicRemappingDetailWrapper.TYPE_REMAPPING, maxEpoch, new HashMap<>(), brokerConfigMap);
+                String oldMappingDataFile = TopicQueueMappingUtils.writeToTemp(oldWrapper, false);
+                System.out.println("The old mapping data is written to file " + oldMappingDataFile);
+            }
+
+            TopicQueueMappingUtils.MappingAllocator allocator = TopicQueueMappingUtils.buildMappingAllocator(new HashMap<>(), targetBrokers.stream().collect(Collectors.toMap( x -> x, x -> 0)));
+            allocator.upToNum(maxNum);
+            Map<String, Integer> expectedBrokerNumMap = allocator.getBrokerNumMap();
+            Queue<Integer> waitAssignQueues = new ArrayDeque<Integer>();
+            Map<Integer, String> expectedIdToBroker = new HashMap<>();
+            //the following logic will make sure that, for one broker, either "map in" or "map out"
+            //It can't both,  map in some queues but also map out some queues.
+            globalIdMap.forEach((queueId, mappingOne) -> {
+                String leaderBroker = mappingOne.getBname();
+                if (expectedBrokerNumMap.containsKey(leaderBroker)) {
+                    if (expectedBrokerNumMap.get(leaderBroker) > 0) {
+                        expectedIdToBroker.put(queueId, leaderBroker);
+                        expectedBrokerNumMap.put(leaderBroker, expectedBrokerNumMap.get(leaderBroker) - 1);
+                    } else {
+                        waitAssignQueues.add(queueId);
+                        expectedBrokerNumMap.remove(leaderBroker);
+                    }
+                } else {
+                    waitAssignQueues.add(queueId);
+                }
+            });
+            expectedBrokerNumMap.forEach((broker, queueNum) -> {
+                for (int i = 0; i < queueNum; i++) {
+                    Integer queueId = waitAssignQueues.poll();
+                    assert queueId != null;
+                    expectedIdToBroker.put(queueId, broker);
+                }
+            });
+            long newEpoch = Math.max(maxEpochAndNum.getKey() + 1000, System.currentTimeMillis());
+
+            //Now construct the remapping info
+            Set<String> brokersToMapOut = new HashSet<>();
+            Set<String> brokersToMapIn = new HashSet<>();
+            for (Map.Entry<Integer, String> mapEntry : expectedIdToBroker.entrySet()) {
+                Integer queueId = mapEntry.getKey();
+                String broker = mapEntry.getValue();
+                TopicQueueMappingOne topicQueueMappingOne = globalIdMap.get(queueId);
+                assert topicQueueMappingOne != null;
+                if (topicQueueMappingOne.getBname().equals(broker)) {
+                    continue;
+                }
+                //remapping
+                String mapInBroker = broker;
+                String mapOutBroker = topicQueueMappingOne.getBname();
+                brokersToMapIn.add(mapInBroker);
+                brokersToMapOut.add(mapOutBroker);
+                TopicConfigAndQueueMapping mapInConfig = brokerConfigMap.get(mapInBroker);
+                TopicConfigAndQueueMapping mapOutConfig = brokerConfigMap.get(mapOutBroker);
+
+                mapInConfig.setWriteQueueNums(mapInConfig.getWriteQueueNums() + 1);
+                mapInConfig.setWriteQueueNums(mapInConfig.getWriteQueueNums() + 1);
+
+                List<LogicQueueMappingItem> items = new ArrayList<>(topicQueueMappingOne.getItems());
+                LogicQueueMappingItem last = items.get(items.size() - 1);
+                items.add(new LogicQueueMappingItem(last.getGen() + 1, mapInConfig.getWriteQueueNums() - 1, mapInBroker, -1, 0, -1, -1, -1));
+
+                ImmutableList<LogicQueueMappingItem> resultItems = ImmutableList.copyOf(items);
+                //Use the same object
+                mapInConfig.getMappingDetail().putMappingInfo(queueId, resultItems);
+                mapOutConfig.getMappingDetail().putMappingInfo(queueId, resultItems);
+            }
+
+            brokerConfigMap.values().forEach(configMapping -> {
+                configMapping.getMappingDetail().setEpoch(newEpoch);
+                configMapping.getMappingDetail().setTotalQueues(maxNum);
+            });
+            //double check
+            TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, brokerConfigMap);
+            TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(brokerConfigMap.values().stream().map(TopicConfigAndQueueMapping::getMappingDetail).collect(Collectors.toList())), false, true);
+
+            {
+                TopicRemappingDetailWrapper newWrapper = new TopicRemappingDetailWrapper(topic, TopicRemappingDetailWrapper.TYPE_REMAPPING, newEpoch, expectedIdToBroker, brokerConfigMap);
+                newWrapper.setBrokerToMapIn(brokersToMapIn);
+                newWrapper.setBrokerToMapOut(brokersToMapOut);
+                String newMappingDataFile = TopicQueueMappingUtils.writeToTemp(newWrapper, true);
+                System.out.println("The old mapping data is written to file " + newMappingDataFile);
+            }
+
+            doRemapping(topic, brokersToMapIn, brokersToMapOut, brokerConfigMap, clientMetadata, defaultMQAdminExt);
+
+        } catch (Exception e) {
+            throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
+        } finally {
+            defaultMQAdminExt.shutdown();
+        }
+    }
+}

tools/src/main/java/org/apache/rocketmq/tools/command/topic/UpdateStaticTopicSubCommand.java

@@ -21,15 +21,18 @@ import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.OptionGroup;
 import org.apache.commons.cli.Options;
-import org.apache.rocketmq.common.LogicQueueMappingItem;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.statictopic.LogicQueueMappingItem;
 import org.apache.rocketmq.common.TopicConfig;
-import org.apache.rocketmq.common.TopicConfigAndQueueMapping;
-import org.apache.rocketmq.common.TopicQueueMappingDetail;
-import org.apache.rocketmq.common.TopicQueueMappingUtils;
+import org.apache.rocketmq.common.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingDetail;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingOne;
+import org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils;
 import org.apache.rocketmq.common.protocol.body.ClusterInfo;
 import org.apache.rocketmq.common.protocol.route.QueueData;
 import org.apache.rocketmq.common.protocol.route.TopicRouteData;
 import org.apache.rocketmq.common.rpc.ClientMetadata;
+import org.apache.rocketmq.common.statictopic.TopicRemappingDetailWrapper;
 import org.apache.rocketmq.remoting.RPCHook;
 import org.apache.rocketmq.srvutil.ServerUtil;
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
@@ -40,7 +43,6 @@ import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -63,7 +65,10 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
 
         Option opt = null;
 
-        opt = new Option("c", "clusterName", true, "create topic to which cluster");
+        opt = new Option("c", "clusters", true, "remapping static topic to clusters, comma separated");
+        optionGroup.addOption(opt);
+
+        opt = new Option("b", "brokers", true, "remapping static topic to brokers, comma separated");
         optionGroup.addOption(opt);
 
         optionGroup.setRequired(true);
@@ -77,104 +82,141 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
         opt.setRequired(true);
         options.addOption(opt);
 
+        opt = new Option("f", "mapFile", true, "The map file name");
+        opt.setRequired(false);
+        options.addOption(opt);
+
         return options;
     }
 
 
+
+    public void executeFromFile(final CommandLine commandLine, final Options options,
+                        RPCHook rpcHook) throws SubCommandException {
+        DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
+        defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
+        ClientMetadata clientMetadata = new ClientMetadata();
+
+        try {
+            String topic = commandLine.getOptionValue('t').trim();
+            String mapFileName = commandLine.getOptionValue('f').trim();
+            String mapData = MixAll.file2String(mapFileName);
+            TopicRemappingDetailWrapper wrapper = TopicRemappingDetailWrapper.decode(mapData.getBytes(), TopicRemappingDetailWrapper.class);
+            //double check the config
+            TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, wrapper.getBrokerConfigMap());
+            TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(wrapper.getBrokerConfigMap().values())), false, true);
+
+            ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
+            if (clusterInfo == null
+                    || clusterInfo.getClusterAddrTable().isEmpty()) {
+                throw new RuntimeException("The Cluster info is empty");
+            }
+            clientMetadata.refreshClusterInfo(clusterInfo);
+            doUpdate(wrapper.getBrokerConfigMap(), clientMetadata, defaultMQAdminExt);
+            return;
+        }catch (Exception e) {
+            throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
+        } finally {
+            defaultMQAdminExt.shutdown();
+        }
+    }
+
+    public void doUpdate(Map<String, TopicConfigAndQueueMapping> brokerConfigMap, ClientMetadata clientMetadata, DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+        //If some succeed, and others fail, it will cause inconsistent data
+        for (Map.Entry<String, TopicConfigAndQueueMapping> entry : brokerConfigMap.entrySet()) {
+            String broker = entry.getKey();
+            String addr = clientMetadata.findMasterBrokerAddr(broker);
+            TopicConfigAndQueueMapping configMapping = entry.getValue();
+            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail());
+        }
+    }
+
+
     @Override
     public void execute(final CommandLine commandLine, final Options options,
         RPCHook rpcHook) throws SubCommandException {
+        if (!commandLine.hasOption('t')) {
+            ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
+            return;
+        }
+
+        if (commandLine.hasOption("f")) {
+            executeFromFile(commandLine, options, rpcHook);
+            return;
+        }
+
         DefaultMQAdminExt defaultMQAdminExt = new DefaultMQAdminExt(rpcHook);
         defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
-
         ClientMetadata clientMetadata = new ClientMetadata();
-        Map<String, TopicConfigAndQueueMapping> existedTopicConfigMap = new HashMap<>();
-        Map<Integer, ImmutableList<LogicQueueMappingItem>> globalIdMap = new HashMap<>();
+
+        Map<String, TopicConfigAndQueueMapping> brokerConfigMap = new HashMap<>();
+        Map<Integer, TopicQueueMappingOne> globalIdMap = new HashMap<>();
+        Set<String> targetBrokers = new HashSet<>();
+
         try {
-            if (!commandLine.hasOption('t')
-                    || !commandLine.hasOption('c')
+            if ((!commandLine.hasOption("b") && !commandLine.hasOption('c'))
                     || !commandLine.hasOption("qn")) {
                 ServerUtil.printCommandLineHelp("mqadmin " + this.commandName(), options);
                 return;
             }
             String topic = commandLine.getOptionValue('t').trim();
-            int queueNum = Integer.parseInt(commandLine.getOptionValue("qn").trim());
-            String clusters = commandLine.getOptionValue('c').trim();
+
             ClusterInfo clusterInfo  = defaultMQAdminExt.examineBrokerClusterInfo();
             if (clusterInfo == null
                     || clusterInfo.getClusterAddrTable().isEmpty()) {
                 throw new RuntimeException("The Cluster info is empty");
-            } else {
-                clientMetadata.refreshClusterInfo(clusterInfo);
             }
-            Set<String> brokers = new HashSet<>();
-            for (String cluster : clusters.split(",")) {
-                cluster = cluster.trim();
-                if (clusterInfo.getClusterAddrTable().get(cluster) != null) {
-                    brokers.addAll(clusterInfo.getClusterAddrTable().get(cluster));
+            clientMetadata.refreshClusterInfo(clusterInfo);
+            {
+                if (commandLine.hasOption("b")) {
+                    String brokerStrs = commandLine.getOptionValue("b").trim();
+                    for (String broker: brokerStrs.split(",")) {
+                        targetBrokers.add(broker.trim());
+                    }
+                } else if (commandLine.hasOption("c")) {
+                    String clusters = commandLine.getOptionValue('c').trim();
+                    for (String cluster : clusters.split(",")) {
+                        cluster = cluster.trim();
+                        if (clusterInfo.getClusterAddrTable().get(cluster) != null) {
+                            targetBrokers.addAll(clusterInfo.getClusterAddrTable().get(cluster));
+                        }
+                    }
+                }
+                if (targetBrokers.isEmpty()) {
+                    throw new RuntimeException("Find none brokers, do nothing");
+                }
+                for (String broker : targetBrokers) {
+                    String addr = clientMetadata.findMasterBrokerAddr(broker);
+                    if (addr == null) {
+                        throw new RuntimeException("Can't find addr for broker " + broker);
+                    }
                 }
-            }
-            if (brokers.isEmpty()) {
-                throw new RuntimeException("Find none brokers for " + clusters);
             }
 
             //get the existed topic config and mapping
-            TopicRouteData routeData = defaultMQAdminExt.examineTopicRouteInfo(topic);
-            clientMetadata.freshTopicRoute(topic, routeData);
-            if (routeData != null
-                    && !routeData.getQueueDatas().isEmpty()) {
-                for (QueueData queueData: routeData.getQueueDatas()) {
-                    String bname = queueData.getBrokerName();
-                    String addr = clientMetadata.findMasterBrokerAddr(bname);
-                    TopicConfigAndQueueMapping mapping = (TopicConfigAndQueueMapping) defaultMQAdminExt.examineTopicConfig(addr, topic);
-                    //allow the config is null
-                    if (mapping != null) {
-                        existedTopicConfigMap.put(bname, mapping);
+            int queueNum = Integer.parseInt(commandLine.getOptionValue("qn").trim());
+            {
+                TopicRouteData routeData = defaultMQAdminExt.examineTopicRouteInfo(topic);
+                clientMetadata.freshTopicRoute(topic, routeData);
+
+                if (routeData != null
+                        && !routeData.getQueueDatas().isEmpty()) {
+                    for (QueueData queueData: routeData.getQueueDatas()) {
+                        String bname = queueData.getBrokerName();
+                        String addr = clientMetadata.findMasterBrokerAddr(bname);
+                        TopicConfigAndQueueMapping mapping = (TopicConfigAndQueueMapping) defaultMQAdminExt.examineTopicConfig(addr, topic);
+                        //allow the config is null
+                        if (mapping != null) {
+                            brokerConfigMap.put(bname, mapping);
+                        }
                     }
                 }
             }
 
             Map.Entry<Long, Integer> maxEpochAndNum = new AbstractMap.SimpleImmutableEntry<>(System.currentTimeMillis(), queueNum);
-            if (!existedTopicConfigMap.isEmpty()) {
-                //make sure it it not null
-                existedTopicConfigMap.forEach((key, value) -> {
-                    if (value.getMappingDetail() != null) {
-                        throw new RuntimeException("Mapping info should be null in broker " + key);
-                    }
-                });
-                //make sure the detail is not dirty
-                existedTopicConfigMap.forEach((key, value) -> {
-                    if (!key.equals(value.getMappingDetail().getBname())) {
-                        throw new RuntimeException(String.format("The broker name is not equal %s != %s ", key, value.getMappingDetail().getBname()));
-                    }
-                    if (value.getMappingDetail().isDirty()) {
-                        throw new RuntimeException("The mapping info is dirty in broker  " + value.getMappingDetail().getBname());
-                    }
-                });
-
-                List<TopicQueueMappingDetail> detailList = existedTopicConfigMap.values().stream().map(TopicConfigAndQueueMapping::getMappingDetail).collect(Collectors.toList());
-                //check the epoch and qnum
-                maxEpochAndNum = TopicQueueMappingUtils.findMaxEpochAndQueueNum(detailList);
-                final Map.Entry<Long, Integer> tmpMaxEpochAndNum = maxEpochAndNum;
-                detailList.forEach( mappingDetail -> {
-                    if (tmpMaxEpochAndNum.getKey() != mappingDetail.getEpoch()) {
-                        throw new RuntimeException(String.format("epoch dose not match %d != %d in %s", tmpMaxEpochAndNum.getKey(), mappingDetail.getEpoch(), mappingDetail.getBname()));
-                    }
-                    if (tmpMaxEpochAndNum.getValue() != mappingDetail.getTotalQueues()) {
-                        throw new RuntimeException(String.format("total queue number dose not match %d != %d in %s", tmpMaxEpochAndNum.getValue(), mappingDetail.getTotalQueues(), mappingDetail.getBname()));
-                    }
-                });
-
-                globalIdMap = TopicQueueMappingUtils.buildMappingItems(new ArrayList<>(detailList), false);
-
-                if (maxEpochAndNum.getValue() != globalIdMap.size()) {
-                    throw new RuntimeException(String.format("The total queue number in config dose not match the real hosted queues %d != %d", maxEpochAndNum.getValue(), globalIdMap.size()));
-                }
-                for (int i = 0; i < maxEpochAndNum.getValue(); i++) {
-                    if (!globalIdMap.containsKey(i)) {
-                        throw new RuntimeException(String.format("The queue number %s is not in globalIdMap", i));
-                    }
-                }
+            if (!brokerConfigMap.isEmpty()) {
+                maxEpochAndNum = TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, brokerConfigMap);
+                globalIdMap = TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
             }
             if (queueNum < globalIdMap.size()) {
                 throw new RuntimeException(String.format("Cannot decrease the queue num for static topic %d < %d", queueNum, globalIdMap.size()));
@@ -183,48 +225,71 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
             if (queueNum == globalIdMap.size()) {
                 throw new RuntimeException("The topic queue num is equal the existed queue num, do nothing");
             }
+
+            {
+                TopicRemappingDetailWrapper oldWrapper = new TopicRemappingDetailWrapper(topic, TopicRemappingDetailWrapper.TYPE_CREATE_OR_UPDATE, maxEpochAndNum.getKey(), new HashMap<>(), brokerConfigMap);
+                String oldMappingDataFile = TopicQueueMappingUtils.writeToTemp(oldWrapper, false);
+                System.out.println("The old mapping data is written to file " + oldMappingDataFile);
+            }
+
+
             //the check is ok, now do the mapping allocation
-            Map<String, Integer> brokerNumMap = brokers.stream().collect(Collectors.toMap( x -> x, x -> 0));
-            Map<Integer, String> idToBroker = new HashMap<>();
+            Map<String, Integer> brokerNumMap = targetBrokers.stream().collect(Collectors.toMap( x -> x, x -> 0));
+            final Map<Integer, String> oldIdToBroker = new HashMap<>();
             globalIdMap.forEach((key, value) -> {
-                String leaderbroker = TopicQueueMappingUtils.getLeaderBroker(value);
-                idToBroker.put(key, leaderbroker);
+                String leaderbroker = value.getBname();
+                oldIdToBroker.put(key, leaderbroker);
                 if (!brokerNumMap.containsKey(leaderbroker)) {
                     brokerNumMap.put(leaderbroker, 1);
                 } else {
                     brokerNumMap.put(leaderbroker, brokerNumMap.get(leaderbroker) + 1);
                 }
             });
-            TopicQueueMappingUtils.MappingAllocator allocator = TopicQueueMappingUtils.buildMappingAllocator(idToBroker, brokerNumMap);
+            TopicQueueMappingUtils.MappingAllocator allocator = TopicQueueMappingUtils.buildMappingAllocator(oldIdToBroker, brokerNumMap);
             allocator.upToNum(queueNum);
             Map<Integer, String> newIdToBroker = allocator.getIdToBroker();
 
             //construct the topic configAndMapping
-            long epoch = Math.max(maxEpochAndNum.getKey() + 1000, System.currentTimeMillis());
-            newIdToBroker.forEach( (queueId, broker) -> {
+            long newEpoch = Math.max(maxEpochAndNum.getKey() + 1000, System.currentTimeMillis());
+            for (Map.Entry<Integer, String> e : newIdToBroker.entrySet()) {
+                Integer queueId = e.getKey();
+                String broker = e.getValue();
+                if (globalIdMap.containsKey(queueId)) {
+                    //ignore the exited
+                    continue;
+                }
                 TopicConfigAndQueueMapping configMapping;
-                if (!existedTopicConfigMap.containsKey(broker)) {
-                    TopicConfig topicConfig = new TopicConfig(topic, 1, 1);
-                    TopicQueueMappingDetail mappingDetail = new TopicQueueMappingDetail(topic, queueNum, broker, epoch);
-                    configMapping = new TopicConfigAndQueueMapping(topicConfig, mappingDetail);
+                if (!brokerConfigMap.containsKey(broker)) {
+                    configMapping = new TopicConfigAndQueueMapping(new TopicConfig(topic), new TopicQueueMappingDetail(topic, 0, broker, -1));
+                    configMapping.setWriteQueueNums(1);
+                    configMapping.setReadQueueNums(1);
+                    brokerConfigMap.put(broker, configMapping);
                 } else {
-                    configMapping = existedTopicConfigMap.get(broker);
+                    configMapping = brokerConfigMap.get(broker);
                     configMapping.setWriteQueueNums(configMapping.getWriteQueueNums() + 1);
-                    configMapping.setWriteQueueNums(configMapping.getWriteQueueNums() + 1);
-                    configMapping.getMappingDetail().setEpoch(epoch);
-                    configMapping.getMappingDetail().setTotalQueues(queueNum);
+                    configMapping.setReadQueueNums(configMapping.getReadQueueNums() + 1);
                 }
                 LogicQueueMappingItem mappingItem = new LogicQueueMappingItem(0, configMapping.getWriteQueueNums() - 1, broker, 0, 0, -1, -1, -1);
                 configMapping.getMappingDetail().putMappingInfo(queueId, ImmutableList.of(mappingItem));
+            }
+
+            // set the topic config
+            brokerConfigMap.values().forEach(configMapping -> {
+                configMapping.getMappingDetail().setEpoch(newEpoch);
+                configMapping.getMappingDetail().setTotalQueues(queueNum);
             });
+            //double check the config
+            TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, brokerConfigMap);
+            TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
 
-            //If some succeed, and others fail, it will cause inconsistent data
-            for (Map.Entry<String, TopicConfigAndQueueMapping> entry : existedTopicConfigMap.entrySet()) {
-                String broker = entry.getKey();
-                String addr = clientMetadata.findMasterBrokerAddr(broker);
-                TopicConfigAndQueueMapping configMapping = entry.getValue();
-                defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail());
+            {
+                TopicRemappingDetailWrapper newWrapper = new TopicRemappingDetailWrapper(topic, TopicRemappingDetailWrapper.TYPE_CREATE_OR_UPDATE, newEpoch, newIdToBroker, brokerConfigMap);
+                String newMappingDataFile = TopicQueueMappingUtils.writeToTemp(newWrapper, true);
+                System.out.println("The new mapping data is written to file " + newMappingDataFile);
             }
+
+            doUpdate(brokerConfigMap, clientMetadata, defaultMQAdminExt);
+
         } catch (Exception e) {
             throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
         } finally {