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

@@ -343,7 +343,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
             this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion());
             response.setCode(ResponseCode.SUCCESS);
         } catch (Exception e) {
-            log.error("Update static failed for [{}]", request, e);
+            log.error("Update static topic failed for [{}]", request, e);
             response.setCode(ResponseCode.SYSTEM_ERROR);
             response.setRemark(e.getMessage());
         }
@@ -653,7 +653,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
                     assert i ==  mappingItems.size() - 1;
                     offset = this.brokerController.getMessageStore().getOffsetInQueueByTime(mappingContext.getTopic(), item.getQueueId(), timestamp);
                     if (offset > 0) {
-                        offset = item.computeStaticQueueOffset(offset);
+                        offset = item.computeStaticQueueOffsetUpToEnd(offset);
                     }
                 } else {
                     requestHeader.setPhysical(true);
@@ -670,7 +670,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
                             || (item.checkIfEndOffsetDecided() && offsetResponseHeader.getOffset() >= item.getEndOffset())) {
                         continue;
                     } else {
-                        offset = item.computeStaticQueueOffset(offsetResponseHeader.getOffset());
+                        offset = item.computeStaticQueueOffsetUpToEnd(offsetResponseHeader.getOffset());
                     }
 
                 }
@@ -722,7 +722,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         }
         long offset = this.brokerController.getMessageStore().getMaxOffsetInQueue(mappingContext.getTopic(), mappingItem.getQueueId());
 
-        offset = mappingItem.computeStaticQueueOffset(offset);
+        offset = mappingItem.computeStaticQueueOffsetUpToEnd(offset);
 
         final RemotingCommand response = RemotingCommand.createResponseCommand(GetMaxOffsetResponseHeader.class);
         final GetMaxOffsetResponseHeader responseHeader = (GetMaxOffsetResponseHeader) response.readCustomHeader();
@@ -774,7 +774,7 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
                 throw rpcResponse.getException();
             }
             GetMinOffsetResponseHeader offsetResponseHeader = (GetMinOffsetResponseHeader) rpcResponse.getHeader();
-            long offset = mappingItem.computeStaticQueueOffset(offsetResponseHeader.getOffset());
+            long offset = mappingItem.computeStaticQueueOffsetUpToEnd(offsetResponseHeader.getOffset());
 
             final RemotingCommand response = RemotingCommand.createResponseCommand(GetMinOffsetResponseHeader.class);
             final GetMinOffsetResponseHeader responseHeader = (GetMinOffsetResponseHeader) response.readCustomHeader();

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

@@ -180,16 +180,16 @@ public class PullMessageProcessor extends AsyncNettyRequestProcessor implements
                         && nextBeginOffset >= mappingItem.getEndOffset()) {
                     nextBeginOffset = mappingItem.getEndOffset();
                 }
-                responseHeader.setNextBeginOffset(mappingItem.computeStaticQueueOffset(nextBeginOffset));
+                responseHeader.setNextBeginOffset(mappingItem.computeStaticQueueOffsetUpToEnd(nextBeginOffset));
             }
             //handle min offset
-            responseHeader.setMinOffset(mappingItem.computeStaticQueueOffset(Math.max(mappingItem.getStartOffset(), responseHeader.getMinOffset())));
+            responseHeader.setMinOffset(mappingItem.computeStaticQueueOffsetUpToEnd(Math.max(mappingItem.getStartOffset(), responseHeader.getMinOffset())));
             //handle max offset
             {
                 if (mappingItem.checkIfEndOffsetDecided()) {
                     responseHeader.setMaxOffset(Math.max(mappingItem.computeMaxStaticQueueOffset(), mappingDetail.computeMaxOffsetFromMapping(mappingContext.getGlobalId())));
                 } else {
-                    responseHeader.setMaxOffset(mappingItem.computeStaticQueueOffset(responseHeader.getMaxOffset()));
+                    responseHeader.setMaxOffset(mappingItem.computeStaticQueueOffsetUpToEnd(responseHeader.getMaxOffset()));
                 }
             }
             //set the offsetDelta

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

@@ -135,7 +135,7 @@ public class SendMessageProcessor extends AbstractSendMessageProcessor implement
                 return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d does not exit in request process of current broker %s", mappingContext.getTopic(), mappingContext.getGlobalId(), mappingDetail.getBname()));
             }
             //no need to care the broker name
-            long staticLogicOffset = mappingItem.computeStaticQueueOffset(responseHeader.getQueueOffset());
+            long staticLogicOffset = mappingItem.computeStaticQueueOffsetUpToEnd(responseHeader.getQueueOffset());
             if (staticLogicOffset < 0) {
                 return buildErrorResponse(ResponseCode.NOT_LEADER_FOR_QUEUE, String.format("%s-%d convert offset error in current broker %s", mappingContext.getTopic(), mappingContext.getGlobalId(), mappingDetail.getBname()));
             }

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

@@ -36,6 +36,7 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -57,9 +58,15 @@ public class TopicQueueMappingManager extends ConfigManager {
         this.brokerController = brokerController;
     }
 
-    public void updateTopicQueueMapping(TopicQueueMappingDetail newDetail, boolean force) {
-        lock.lock();
+    public void updateTopicQueueMapping(TopicQueueMappingDetail newDetail, boolean force) throws Exception {
+        boolean locked = false;
+        boolean updated = false;
         try {
+            if (lock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
+                locked = true;
+            } else {
+                return;
+            }
             if (newDetail == null) {
                 return;
             }
@@ -70,6 +77,7 @@ public class TopicQueueMappingManager extends ConfigManager {
             TopicQueueMappingDetail oldDetail = topicQueueMappingTable.get(newDetail.getTopic());
             if (oldDetail == null) {
                 topicQueueMappingTable.put(newDetail.getTopic(), newDetail);
+                updated = true;
                 return;
             }
             if (force) {
@@ -77,6 +85,7 @@ public class TopicQueueMappingManager extends ConfigManager {
                     newDetail.getHostedQueues().putIfAbsent(queueId, items);
                 });
                 topicQueueMappingTable.put(newDetail.getTopic(), newDetail);
+                updated = true;
                 return;
             }
             //do more check
@@ -94,8 +103,14 @@ public class TopicQueueMappingManager extends ConfigManager {
                 }
             }
             topicQueueMappingTable.put(newDetail.getTopic(), newDetail);
+            updated = true;
         }  finally {
-            lock.unlock();
+            if (locked) {
+                this.lock.unlock();
+            }
+            if (updated) {
+                this.persist();
+            }
         }
 
     }

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

@@ -10,8 +10,11 @@ import org.apache.rocketmq.common.protocol.route.TopicRouteData;
 import org.apache.rocketmq.logging.InternalLogger;
 import org.apache.rocketmq.logging.InternalLoggerFactory;
 
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -93,33 +96,46 @@ public class ClientMetadata {
                 || route.getTopicQueueMappingByBroker().isEmpty()) {
             return new ConcurrentHashMap<MessageQueue, String>();
         }
-        ConcurrentMap<MessageQueue, String> mqEndPoints = new ConcurrentHashMap<MessageQueue, String>();
+        ConcurrentMap<MessageQueue, TopicQueueMappingInfo> mqEndPoints = new ConcurrentHashMap<MessageQueue, TopicQueueMappingInfo>();
 
-        int totalNums = 0;
-        for (Map.Entry<String, TopicQueueMappingInfo> entry : route.getTopicQueueMappingByBroker().entrySet()) {
-            String brokerName = entry.getKey();
-            //TODO check the epoch of
-            if (entry.getValue().getTotalQueues() > totalNums) {
-                if (totalNums != 0) {
-                    log.warn("The static logic queue totalNums dose not match before {} {} != {}", topic, totalNums, entry.getValue().getTotalQueues());
+
+        List<Map.Entry<String, TopicQueueMappingInfo>> mappingInfos = new ArrayList<Map.Entry<String, TopicQueueMappingInfo>>(route.getTopicQueueMappingByBroker().entrySet());
+        Collections.sort(mappingInfos, new Comparator<Map.Entry<String, TopicQueueMappingInfo>>() {
+            @Override
+            public int compare(Map.Entry<String, TopicQueueMappingInfo> o1, Map.Entry<String, TopicQueueMappingInfo> o2) {
+                return  (int) (o2.getValue().getEpoch() - o1.getValue().getEpoch());
             }
-                totalNums = entry.getValue().getTotalQueues();
+        });
+
+        int maxTotalNums = 0;
+        long maxTotalNumOfEpoch = -1;
+        for (Map.Entry<String, TopicQueueMappingInfo> entry : mappingInfos) {
+            TopicQueueMappingInfo info = entry.getValue();
+            if (info.getEpoch() >= maxTotalNumOfEpoch && info.getTotalQueues() > maxTotalNums) {
+                maxTotalNums = entry.getValue().getTotalQueues();
             }
             for (Map.Entry<Integer, Integer> idEntry : entry.getValue().getCurrIdMap().entrySet()) {
                 int globalId = idEntry.getKey();
                 MessageQueue mq = new MessageQueue(topic, MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME, globalId);
-                String oldBrokerName = mqEndPoints.put(mq, brokerName);
-                log.warn("The static logic queue is duplicated {} {} {} ", mq, oldBrokerName, brokerName);
+                TopicQueueMappingInfo oldInfo = mqEndPoints.get(mq);
+                if (oldInfo == null ||  oldInfo.getEpoch() <= info.getEpoch()) {
+                    mqEndPoints.put(mq, info);
+                }
             }
         }
+
+        ConcurrentMap<MessageQueue, String> mqEndPointsOfBroker = new ConcurrentHashMap<MessageQueue, String>();
+
         //accomplish the static logic queues
-        for (int i = 0; i < totalNums; i++) {
+        for (int i = 0; i < maxTotalNums; i++) {
             MessageQueue mq = new MessageQueue(topic, MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME, i);
             if (!mqEndPoints.containsKey(mq)) {
-                mqEndPoints.put(mq, MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME_NOT_EXIST);
+                mqEndPointsOfBroker.put(mq, MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME_NOT_EXIST);
+            } else {
+                mqEndPointsOfBroker.put(mq, mqEndPoints.get(mq).getBname());
             }
         }
-        return mqEndPoints;
+        return mqEndPointsOfBroker;
     }
 
 }

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

@@ -2,14 +2,14 @@ package org.apache.rocketmq.common.statictopic;
 
 public class LogicQueueMappingItem {
 
-    private int gen; //generation, mutable
-    private int queueId;
-    private String bname;
-    private long logicOffset; // the start of the logic offset
-    private long startOffset; // the start of the physical offset, included
-    private long endOffset = -1; // the end of the physical offset, excluded
-    private long timeOfStart = -1; // mutable
-    private long timeOfEnd = -1; // mutable
+    private final int gen; // immutable
+    private final int queueId; //, immutable
+    private final String bname; //important, immutable
+    private long logicOffset; // the start of the logic offset, important, can be changed by command only once
+    private final long startOffset; // the start of the physical offset, should always be 0, immutable
+    private long endOffset = -1; // the end of the physical offset, excluded, revered -1, mutable
+    private long timeOfStart = -1; // mutable, reserved
+    private long timeOfEnd = -1; // mutable, reserved
 
     public LogicQueueMappingItem(int gen, int queueId, String bname, long logicOffset, long startOffset, long endOffset, long timeOfStart, long timeOfEnd) {
         this.gen = gen;
@@ -22,7 +22,7 @@ public class LogicQueueMappingItem {
         this.timeOfEnd = timeOfEnd;
     }
 
-    public long computeStaticQueueOffset(long physicalQueueOffset) {
+    public long computeStaticQueueOffsetUpToEnd(long physicalQueueOffset) {
         if (physicalQueueOffset < startOffset) {
             return logicOffset;
         }
@@ -33,6 +33,13 @@ public class LogicQueueMappingItem {
         return  logicOffset + (physicalQueueOffset - startOffset);
     }
 
+    public long computeStaticQueueOffset(long physicalQueueOffset) {
+        if (physicalQueueOffset < startOffset) {
+            return logicOffset;
+        }
+        return  logicOffset + (physicalQueueOffset - startOffset);
+    }
+
     public long computePhysicalQueueOffset(long staticQueueOffset) {
         return  (staticQueueOffset - logicOffset) + startOffset;
     }

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

@@ -45,14 +45,13 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
 
     public void buildIdMap() {
         this.currIdMap = buildIdMap(LEVEL_0);
-        this.prevIdMap = buildIdMap(LEVEL_1);
     }
 
 
     public ConcurrentMap<Integer, Integer> buildIdMap(int level) {
         //level 0 means current leader in this broker
-        //level 1 means previous leader in this broker
-        assert level == LEVEL_0 || level == LEVEL_1;
+        //level 1 means previous leader in this broker, reserved for
+        assert level == LEVEL_0 ;
 
         if (hostedQueues == null || hostedQueues.isEmpty()) {
             return new ConcurrentHashMap<Integer, Integer>();
@@ -67,12 +66,6 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
                 if (bname.equals(curr.getBname())) {
                     tmpIdMap.put(globalId, curr.getQueueId());
                 }
-            } else if (level == LEVEL_1
-                    && items.size() >= 2) {
-                LogicQueueMappingItem prev = items.get(items.size() - 1);
-                if (bname.equals(prev.getBname())) {
-                    tmpIdMap.put(globalId, prev.getQueueId());
-                }
             }
         }
         return tmpIdMap;
@@ -120,8 +113,6 @@ public class TopicQueueMappingDetail extends TopicQueueMappingInfo {
     public TopicQueueMappingInfo cloneAsMappingInfo() {
         TopicQueueMappingInfo topicQueueMappingInfo = new TopicQueueMappingInfo(this.topic, this.totalQueues, this.bname, this.epoch);
         topicQueueMappingInfo.currIdMap = this.buildIdMap(LEVEL_0);
-        topicQueueMappingInfo.prevIdMap = this.buildIdMap(LEVEL_1);
-
         return topicQueueMappingInfo;
     }
 

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

@@ -23,7 +23,6 @@ import java.util.concurrent.ConcurrentMap;
 
 public class TopicQueueMappingInfo extends RemotingSerializable {
     public static final int LEVEL_0 = 0;
-    public static final int LEVEL_1 = 1;
 
     String topic; // redundant field
     int totalQueues;
@@ -32,8 +31,6 @@ public class TopicQueueMappingInfo extends RemotingSerializable {
     boolean dirty; //indicate if the data is dirty
     //register to broker to construct the route
     transient ConcurrentMap<Integer/*logicId*/, Integer/*physicalId*/> currIdMap = new ConcurrentHashMap<Integer, Integer>();
-    //register to broker to help detect remapping failure
-    transient ConcurrentMap<Integer/*logicId*/, Integer/*physicalId*/> prevIdMap = new ConcurrentHashMap<Integer, Integer>();
 
     public TopicQueueMappingInfo(String topic, int totalQueues, String bname, long epoch) {
         this.topic = topic;
@@ -79,8 +76,4 @@ public class TopicQueueMappingInfo extends RemotingSerializable {
     public ConcurrentMap<Integer, Integer> getCurrIdMap() {
         return currIdMap;
     }
-
-    public ConcurrentMap<Integer, Integer> getPrevIdMap() {
-        return prevIdMap;
-    }
 }

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

@@ -296,6 +296,16 @@ public class TopicQueueMappingUtils {
         }
     }
 
+    public static long blockSeqRoundUp(long offset, long blockSeqSize) {
+        long num = offset/blockSeqSize;
+        long left = offset % blockSeqSize;
+        if (left < blockSeqSize/2) {
+            return (num + 1) * blockSeqSize;
+        } else {
+            return (num + 2) * blockSeqSize;
+        }
+    }
+
 
 
 }

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

@@ -168,6 +168,7 @@ public class RouteInfoManager {
                             if (!topicQueueMappingInfoTable.containsKey(entry.getKey())) {
                                 topicQueueMappingInfoTable.put(entry.getKey(), new HashMap<String, TopicQueueMappingInfo>());
                             }
+                            //Note asset brokerName equal entry.getValue().getBname()
                             topicQueueMappingInfoTable.get(entry.getKey()).put(entry.getValue().getBname(), entry.getValue());
                         }
                     }

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

@@ -81,9 +81,14 @@ public class RemappingStaticTopicSubCommand implements SubCommand {
         opt.setRequired(true);
         options.addOption(opt);
 
-        opt = new Option("f", "mapFile", true, "The map file name");
+        opt = new Option("mf", "mapFile", true, "The mapping data file name ");
         opt.setRequired(false);
         options.addOption(opt);
+
+        opt = new Option("fr", "forceReplace", true, "Force replace the old mapping");
+        opt.setRequired(false);
+        options.addOption(opt);
+
         return options;
     }
 
@@ -111,7 +116,12 @@ public class RemappingStaticTopicSubCommand implements SubCommand {
                 throw new RuntimeException("The Cluster info is empty");
             }
             clientMetadata.refreshClusterInfo(clusterInfo);
-            doRemapping(topic, wrapper.getBrokerToMapIn(), wrapper.getBrokerToMapOut(), wrapper.getBrokerConfigMap(), clientMetadata, defaultMQAdminExt, false);
+
+            boolean force = false;
+            if (commandLine.hasOption("fr") && Boolean.parseBoolean(commandLine.getOptionValue("fr").trim())) {
+                force = true;
+            }
+            doRemapping(topic, wrapper.getBrokerToMapIn(), wrapper.getBrokerToMapOut(), wrapper.getBrokerConfigMap(), clientMetadata, defaultMQAdminExt, force);
             return;
         }catch (Exception e) {
             throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
@@ -161,7 +171,7 @@ public class RemappingStaticTopicSubCommand implements SubCommand {
                 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));
+                newLeader.setLogicOffset(TopicQueueMappingUtils.blockSeqRoundUp(oldLeader.computeStaticQueueOffset(topicOffset.getMaxOffset()), 10000));
                 TopicConfigAndQueueMapping mapInConfig = brokerConfigMap.get(newLeader.getBname());
                 //fresh the new leader
                 mapInConfig.getMappingDetail().putMappingInfo(globalId, items);
@@ -171,7 +181,7 @@ public class RemappingStaticTopicSubCommand implements SubCommand {
         for (String broker: brokersToMapIn) {
             String addr = clientMetadata.findMasterBrokerAddr(broker);
             TopicConfigAndQueueMapping configMapping = brokerConfigMap.get(broker);
-            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail(), false);
+            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail(), force);
         }
     }
 

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

@@ -82,7 +82,11 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
         opt.setRequired(true);
         options.addOption(opt);
 
-        opt = new Option("f", "mapFile", true, "The map file name");
+        opt = new Option("mf", "mapFile", true, "The mapping data file name");
+        opt.setRequired(false);
+        options.addOption(opt);
+
+        opt = new Option("fr", "forceReplace", true, "Force replace the old mapping");
         opt.setRequired(false);
         options.addOption(opt);
 
@@ -104,6 +108,10 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
             TopicRemappingDetailWrapper wrapper = TopicRemappingDetailWrapper.decode(mapData.getBytes(), TopicRemappingDetailWrapper.class);
             //double check the config
             TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, wrapper.getBrokerConfigMap());
+            boolean force = false;
+            if (commandLine.hasOption("fr") && Boolean.parseBoolean(commandLine.getOptionValue("fr").trim())) {
+                force = true;
+            }
             TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(wrapper.getBrokerConfigMap().values())), false, true);
 
             ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
@@ -112,7 +120,7 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
                 throw new RuntimeException("The Cluster info is empty");
             }
             clientMetadata.refreshClusterInfo(clusterInfo);
-            doUpdate(wrapper.getBrokerConfigMap(), clientMetadata, defaultMQAdminExt);
+            doUpdate(wrapper.getBrokerConfigMap(), clientMetadata, defaultMQAdminExt, force);
             return;
         }catch (Exception e) {
             throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
@@ -121,13 +129,13 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
         }
     }
 
-    public void doUpdate(Map<String, TopicConfigAndQueueMapping> brokerConfigMap, ClientMetadata clientMetadata, DefaultMQAdminExt defaultMQAdminExt) throws Exception {
+    public void doUpdate(Map<String, TopicConfigAndQueueMapping> brokerConfigMap, ClientMetadata clientMetadata, DefaultMQAdminExt defaultMQAdminExt, boolean force) 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(), false);
+            defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail(), force);
         }
     }
 
@@ -288,7 +296,7 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
                 System.out.println("The new mapping data is written to file " + newMappingDataFile);
             }
 
-            doUpdate(brokerConfigMap, clientMetadata, defaultMQAdminExt);
+            doUpdate(brokerConfigMap, clientMetadata, defaultMQAdminExt, false);
 
         } catch (Exception e) {
             throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);