Polish the code for command

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

@@ -116,13 +116,12 @@ public class TopicQueueMappingUtils {
         return detailList;
     }
 
-    public static Map.Entry<Long, Integer> validConsistenceOfTopicConfigAndQueueMapping(Map<String, TopicConfigAndQueueMapping> brokerConfigMap) {
+    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
-        String topic = null;
         long maxEpoch = -1;
         int maxNum = -1;
         for (Map.Entry<String, TopicConfigAndQueueMapping> entry : brokerConfigMap.entrySet()) {
@@ -143,9 +142,7 @@ public class TopicQueueMappingUtils {
             }
             if (topic != null
                 && !topic.equals(mappingDetail.getTopic())) {
-                throw new RuntimeException("The topic name is inconsistent in broker  " + broker);
-            } else {
-                topic = mappingDetail.getTopic();
+                throw new RuntimeException("The topic name is not match for broker  " + broker);
             }
 
             if (maxEpoch != -1
@@ -165,7 +162,7 @@ public class TopicQueueMappingUtils {
         return new AbstractMap.SimpleEntry<Long, Integer>(maxEpoch, maxNum);
     }
 
-    public static Map<Integer, TopicQueueMappingOne> buildMappingItems(List<TopicQueueMappingDetail> mappingDetailList, boolean replace, boolean checkConsistence) {
+    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) {
@@ -216,10 +213,14 @@ public class TopicQueueMappingUtils {
         return items.get(items.size() - 1);
     }
 
-    public static String writeToTemp(TopicRemappingDetailWrapper wrapper, String suffix) {
+    public static String writeToTemp(TopicRemappingDetailWrapper wrapper, boolean after) {
         String topic = wrapper.getTopic();
         String data = wrapper.toJson();
-        String fileName = System.getProperty("java.io.tmpdir") + File.separator + topic + "-" + wrapper.getEpoch() + "-" + suffix;
+        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;

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

@@ -3,12 +3,17 @@ 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;
@@ -17,6 +22,10 @@ public class TopicRemappingDetailWrapper extends RemotingSerializable {
 
     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;
@@ -44,4 +53,20 @@ public class TopicRemappingDetailWrapper extends RemotingSerializable {
     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;
+    }
 }

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

@@ -39,7 +39,6 @@ import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
 import org.apache.rocketmq.tools.command.SubCommand;
 import org.apache.rocketmq.tools.command.SubCommandException;
 
-import java.nio.charset.Charset;
 import java.util.AbstractMap;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -84,7 +83,7 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
         options.addOption(opt);
 
         opt = new Option("f", "mapFile", true, "The map file name");
-        opt.setRequired(true);
+        opt.setRequired(false);
         options.addOption(opt);
 
         return options;
@@ -99,12 +98,13 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
         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.validConsistenceOfTopicConfigAndQueueMapping(wrapper.getBrokerConfigMap());
-            TopicQueueMappingUtils.buildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(wrapper.getBrokerConfigMap().values())), false, true);
+            TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, wrapper.getBrokerConfigMap());
+            TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(wrapper.getBrokerConfigMap().values())), false, true);
 
             ClusterInfo clusterInfo = defaultMQAdminExt.examineBrokerClusterInfo();
             if (clusterInfo == null
@@ -131,6 +131,11 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
     @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;
@@ -159,7 +164,6 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
             }
             clientMetadata.refreshClusterInfo(clusterInfo);
 
-
             String clusters = commandLine.getOptionValue('c').trim();
             for (String cluster : clusters.split(",")) {
                 cluster = cluster.trim();
@@ -200,8 +204,8 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
 
             Map.Entry<Long, Integer> maxEpochAndNum = new AbstractMap.SimpleImmutableEntry<>(System.currentTimeMillis(), queueNum);
             if (!brokerConfigMap.isEmpty()) {
-                maxEpochAndNum = TopicQueueMappingUtils.validConsistenceOfTopicConfigAndQueueMapping(brokerConfigMap);
-                globalIdMap = TopicQueueMappingUtils.buildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
+                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()));
@@ -213,7 +217,7 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
 
             {
                 TopicRemappingDetailWrapper oldWrapper = new TopicRemappingDetailWrapper(topic, TopicRemappingDetailWrapper.TYPE_CREATE_OR_UPDATE, maxEpochAndNum.getKey(), new HashMap<>(), brokerConfigMap);
-                String oldMappingDataFile = TopicQueueMappingUtils.writeToTemp(oldWrapper, "before");
+                String oldMappingDataFile = TopicQueueMappingUtils.writeToTemp(oldWrapper, false);
                 System.out.println("The old mapping data is written to file " + oldMappingDataFile);
             }
 
@@ -264,12 +268,12 @@ public class UpdateStaticTopicSubCommand implements SubCommand {
                 configMapping.getMappingDetail().setTotalQueues(queueNum);
             });
             //double check the config
-            TopicQueueMappingUtils.validConsistenceOfTopicConfigAndQueueMapping(brokerConfigMap);
-            TopicQueueMappingUtils.buildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
+            TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, brokerConfigMap);
+            TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(TopicQueueMappingUtils.getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
 
             {
                 TopicRemappingDetailWrapper newWrapper = new TopicRemappingDetailWrapper(topic, TopicRemappingDetailWrapper.TYPE_CREATE_OR_UPDATE, newEpoch, newIdToBroker, brokerConfigMap);
-                String newMappingDataFile = TopicQueueMappingUtils.writeToTemp(newWrapper, "after");
+                String newMappingDataFile = TopicQueueMappingUtils.writeToTemp(newWrapper, true);
                 System.out.println("The new mapping data is written to file " + newMappingDataFile);
             }