Add IT Test for static topic

test/src/test/java/org/apache/rocketmq/test/base/BaseConf.java

@@ -19,8 +19,10 @@ package org.apache.rocketmq.test.base;
 
 import com.google.common.collect.ImmutableList;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ForkJoinPool;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -120,6 +122,13 @@ public class BaseConf {
         return group;
     }
 
+    public static DefaultMQAdminExt getAdmin(String nsAddr) {
+        final DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt(500);
+        mqAdminExt.setNamesrvAddr(nsAddr);
+        mqClients.add(mqAdminExt);
+        return mqAdminExt;
+    }
+
     public static RMQNormalProducer getProducer(String nsAddr, String topic) {
         return getProducer(nsAddr, topic, false);
     }
@@ -197,6 +206,13 @@ public class BaseConf {
         shutdown(mqClients);
     }
 
+    public static Set<String> getBrokers() {
+        Set<String> brokers = new HashSet<>();
+        brokers.add(broker1Name);
+        brokers.add(broker2Name);
+        return brokers;
+    }
+
     public static void shutdown(List<Object> mqClients) {
         mqClients.forEach(mqClient -> ForkJoinPool.commonPool().execute(() -> {
             if (mqClient instanceof AbstractMQProducer) {

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

+package org.apache.rocketmq.test.smoke;
+
+import org.apache.log4j.Logger;
+import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.rpc.ClientMetadata;
+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.test.base.BaseConf;
+import org.apache.rocketmq.test.util.MQRandomUtils;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.rocketmq.common.statictopic.TopicQueueMappingUtils.getMappingDetailFromConfig;
+
+@FixMethodOrder
+public class StaticTopicIT extends BaseConf {
+
+    private static Logger logger = Logger.getLogger(StaticTopicIT.class);
+    private DefaultMQAdminExt defaultMQAdminExt;
+    private ClientMetadata clientMetadata;
+
+    @Before
+    public void setUp() throws Exception {
+        defaultMQAdminExt = getAdmin(nsAddr);
+        waitBrokerRegistered(nsAddr, clusterName);
+        clientMetadata = new ClientMetadata();
+        ClusterInfo clusterInfo  = defaultMQAdminExt.examineBrokerClusterInfo();
+        if (clusterInfo == null
+                || clusterInfo.getClusterAddrTable().isEmpty()) {
+            throw new RuntimeException("The Cluster info is empty");
+        }
+        clientMetadata.refreshClusterInfo(clusterInfo);
+
+    }
+
+    @Test
+    public void testCreateStaticTopic() throws Exception {
+        String topic = "static" + MQRandomUtils.getRandomTopic();
+        int queueNum = 10;
+        Set<String> brokers = getBrokers();
+        //create topic
+        {
+            Map<String, TopicConfigAndQueueMapping> brokerConfigMap = defaultMQAdminExt.examineTopicConfigAll(clientMetadata, topic);
+            Assert.assertTrue(brokerConfigMap.isEmpty());
+            TopicQueueMappingUtils.createTopicConfigMapping(topic, queueNum, getBrokers(), brokerConfigMap);
+            //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);
+            }
+        }
+        Map<String, TopicConfigAndQueueMapping> brokerConfigMap = defaultMQAdminExt.examineTopicConfigAll(clientMetadata, topic);
+
+        TopicQueueMappingUtils.checkConsistenceOfTopicConfigAndQueueMapping(topic, brokerConfigMap);
+        Map<Integer, TopicQueueMappingOne>  globalIdMap = TopicQueueMappingUtils.checkAndBuildMappingItems(new ArrayList<>(getMappingDetailFromConfig(brokerConfigMap.values())), false, true);
+        Assert.assertEquals(queueNum, globalIdMap.size());
+
+    }
+
+    @After
+    public void tearDown() {
+        super.shutdown();
+    }
+
+}

tools/src/test/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtTest.java

@@ -150,13 +150,7 @@ public class DefaultMQAdminExtTest {
         topicRouteData.setBrokerDatas(brokerDatas);
         topicRouteData.setQueueDatas(new ArrayList<QueueData>());
         topicRouteData.setFilterServerTable(new HashMap<String, List<String>>());
-        LogicalQueuesInfo logicalQueuesInfoinfo = new LogicalQueuesInfo();
-        logicalQueuesInfoinfo.put(0, Lists.newArrayList(
-            new LogicalQueueRouteData(0, 0, new MessageQueue(topic1, broker1Name, 0), MessageQueueRouteState.ReadOnly, 0, 1000, 2000, 3000, broker1Addr),
-            new LogicalQueueRouteData(0, 1000, new MessageQueue(topic1, broker2Name, 0), MessageQueueRouteState.Normal, 0, -1, -1, -1, broker2Addr)
-        ));
-        topicRouteData.setLogicalQueuesInfo(logicalQueuesInfoinfo);
-        when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong(), anyBoolean(), any())).thenReturn(topicRouteData);
+        when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(topicRouteData);
 
         HashMap<String, String> result = new HashMap<>();
         result.put("id", String.valueOf(MixAll.MASTER_ID));