Add the allocator test

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

@@ -338,9 +338,6 @@ public class AdminBrokerProcessor extends AsyncNettyRequestProcessor implements
         try {
             this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
 
-            System.out.println("Broker body:" + new String(request.getBody()));
-            System.out.println("Broker bodetaildy:" + topicQueueMappingDetail.toJson());
-
             this.brokerController.getTopicQueueMappingManager().updateTopicQueueMapping(topicQueueMappingDetail, force);
 
             this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion());

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

@@ -17,13 +17,12 @@
 
 package org.apache.rocketmq.common.protocol.body;
 
+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/statictopic/TopicQueueMappingUtils.java

@@ -49,11 +49,12 @@ public class TopicQueueMappingUtils {
         }
 
         private void freshState() {
-            int minNum = -1;
+            int minNum = Integer.MAX_VALUE;
             for (Map.Entry<String, Integer> entry : brokerNumMap.entrySet()) {
-                if (entry.getValue() > minNum) {
+                if (entry.getValue() < minNum) {
                     leastBrokers.clear();
                     leastBrokers.add(entry.getKey());
+                    minNum = entry.getValue();
                 } else if (entry.getValue() == minNum) {
                     leastBrokers.add(entry.getKey());
                 }

common/src/test/java/org/apache/rocketmq/common/statictopic/TopicMappingUtilsTest.java

+package org.apache.rocketmq.common.statictopic;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class TopicMappingUtilsTest {
+
+
+    private Map<String, Integer> buildBrokerNumMap(int num) {
+        Map<String, Integer> map = new HashMap<String, Integer>();
+        for (int i = 0; i < num; i++) {
+            map.put("broker" + i, 0);
+        }
+        return map;
+    }
+
+    private void testIdToBroker(Map<Integer, String> idToBroker, Map<String, Integer> brokerNumMap) {
+        Map<String, Integer> brokerNumOther = new HashMap<String, Integer>();
+        for (int i = 0; i < idToBroker.size(); i++) {
+            Assert.assertTrue(idToBroker.containsKey(i));
+            String broker = idToBroker.get(i);
+            if (brokerNumOther.containsKey(broker)) {
+                brokerNumOther.put(broker, brokerNumOther.get(broker) + 1);
+            } else {
+                brokerNumOther.put(broker, 1);
+            }
+        }
+        Assert.assertEquals(brokerNumMap.size(), brokerNumOther.size());
+        for (Map.Entry<String, Integer> entry: brokerNumOther.entrySet()) {
+            Assert.assertEquals(entry.getValue(), brokerNumMap.get(entry.getKey()));
+        }
+    }
+
+    @Test
+    public void testAllocator() {
+        //stability
+        for (int i = 0; i < 10; i++) {
+            int num = 3;
+            Map<String, Integer> brokerNumMap = buildBrokerNumMap(num);
+            TopicQueueMappingUtils.MappingAllocator  allocator = TopicQueueMappingUtils.buildMappingAllocator(new HashMap<Integer, String>(), brokerNumMap);
+            allocator.upToNum(num * 2);
+            for (Map.Entry<String, Integer> entry: allocator.getBrokerNumMap().entrySet()) {
+                Assert.assertEquals(2L, entry.getValue().longValue());
+            }
+            Assert.assertEquals(num * 2, allocator.getIdToBroker().size());
+            testIdToBroker(allocator.idToBroker, allocator.getBrokerNumMap());
+
+            allocator.upToNum(num * 3 - 1);
+
+            for (Map.Entry<String, Integer> entry: allocator.getBrokerNumMap().entrySet()) {
+                Assert.assertTrue(entry.getValue() >= 2);
+                Assert.assertTrue(entry.getValue() <= 3);
+            }
+            Assert.assertEquals(num * 3 - 1, allocator.getIdToBroker().size());
+            testIdToBroker(allocator.idToBroker, allocator.getBrokerNumMap());
+        }
+    }
+}

test/src/test/java/org/apache/rocketmq/test/smoke/StaticTopicIT.java

@@ -57,7 +57,6 @@ public class StaticTopicIT extends BaseConf {
                 String broker = entry.getKey();
                 String addr = clientMetadata.findMasterBrokerAddr(broker);
                 TopicConfigAndQueueMapping configMapping = entry.getValue();
-                System.out.println(configMapping.getMappingDetail().toJson());
                 defaultMQAdminExt.createStaticTopic(addr, defaultMQAdminExt.getCreateTopicKey(), configMapping, configMapping.getMappingDetail(), false);
             }
         }