Merge pull request #254 from ni-ze/develop

[ISSUE #255] keep state topic consistent with source topic

Author: ni-ze

core/src/main/java/org/apache/rocketmq/streams/core/common/Constant.java

@@ -50,4 +50,6 @@ public class Constant {
 
     public static final String WORKER_THREAD_NAME = "worker_thread";
 
+    public static final String STATIC_TOPIC_BROKER_NAME = "__syslo__global__";
+
 }

core/src/main/java/org/apache/rocketmq/streams/core/running/WorkerThread.java

@@ -56,13 +56,17 @@ public class WorkerThread extends Thread {
     private final TopologyBuilder topologyBuilder;
     private final PlanetaryEngine<?, ?> planetaryEngine;
     private final Properties properties;
+    private final String jobId;
 
 
     public WorkerThread(String threadName, TopologyBuilder topologyBuilder, Properties properties) throws MQClientException {
         super(threadName);
+
         this.topologyBuilder = topologyBuilder;
         this.properties = properties;
-        String groupName = topologyBuilder.getJobId() + "_" + ROCKETMQ_STREAMS_CONSUMER_GROUP;
+        jobId = topologyBuilder.getJobId();
+
+        String groupName = String.join("_", jobId, ROCKETMQ_STREAMS_CONSUMER_GROUP);
 
         RocketMQClient rocketMQClient = new RocketMQClient(properties.getProperty(MixAll.NAMESRV_ADDR_PROPERTY));
 
@@ -88,8 +92,9 @@ public WorkerThread(String threadName, TopologyBuilder topologyBuilder, Properti
     public void run() {
         try {
             this.planetaryEngine.start();
-
             this.planetaryEngine.runInLoop();
+            logger.info("worker thread=[{}], start task success, jobId:{}", this.getName(), jobId);
+
         } catch (Throwable e) {
             logger.error("worker thread=[{}], error:{}.", this.getName(), e);
             throw new RStreamsException(e);
@@ -101,6 +106,7 @@ public void run() {
 
     public void shutdown() {
         this.planetaryEngine.stop();
+        logger.info("worker thread=[{}], shutdown task success, jobId:{}", this.getName(), jobId);
     }
 
 

core/src/main/java/org/apache/rocketmq/streams/core/state/RocketMQStore.java

@@ -185,7 +185,8 @@ public void persist(Set<MessageQueue> messageQueues) throws Throwable {
             }
 
             String stateTopic = stateTopicQueue.getTopic();
-            createStateTopic(stateTopic);
+            boolean isStaticTopic = stateTopicQueue.getBrokerName().equals(Constant.STATIC_TOPIC_BROKER_NAME);
+            createStateTopic(stateTopic, isStaticTopic);
 
             for (byte[] key : keySet) {
 
@@ -224,7 +225,7 @@ public void loadState(Set<MessageQueue> addQueues) throws Throwable {
 
         Set<MessageQueue> stateTopicQueue = convertSourceTopicQueue2StateTopicQueue(addQueues);
         for (MessageQueue messageQueue : stateTopicQueue) {
-            createStateTopic(messageQueue.getTopic());
+            createStateTopic(messageQueue.getTopic(), messageQueue.getBrokerName().equals(Constant.STATIC_TOPIC_BROKER_NAME));
         }
 
         consumer.assign(stateTopicQueue);
@@ -389,15 +390,19 @@ private List<MessageExt> sortByQueueOffset(List<MessageExt> target) {
         return target;
     }
 
-    private void createStateTopic(String stateTopic) throws Exception {
+    private void createStateTopic(String stateTopic, boolean sourceTopicIsStaticTopic) throws Exception {
         if (RocketMQUtil.checkWhetherExist(stateTopic)) {
             return;
         }
 
         String sourceTopic = stateTopic2SourceTopic(stateTopic);
         Pair<Integer, Set<String>> clustersPair = getTotalQueueNumAndClusters(sourceTopic);
 
-        RocketMQUtil.createNormalTopic(mqAdmin, stateTopic, clustersPair.getKey(), clustersPair.getValue());
+        if (sourceTopicIsStaticTopic) {
+            RocketMQUtil.createStaticCompactTopic(mqAdmin, stateTopic, clustersPair.getKey(), clustersPair.getValue());
+        } else {
+            RocketMQUtil.createNormalTopic(mqAdmin, sourceTopic, stateTopic);
+        }
     }
 
     private Pair<Integer, Set<String>> getTotalQueueNumAndClusters(String sourceTopic) throws Exception {

core/src/main/java/org/apache/rocketmq/streams/core/util/RocketMQUtil.java

@@ -25,6 +25,9 @@
 import org.apache.rocketmq.common.constant.PermName;
 import org.apache.rocketmq.common.protocol.ResponseCode;
 import org.apache.rocketmq.common.protocol.body.ClusterInfo;
+import org.apache.rocketmq.common.protocol.route.BrokerData;
+import org.apache.rocketmq.common.protocol.route.QueueData;
+import org.apache.rocketmq.common.protocol.route.TopicRouteData;
 import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.apache.rocketmq.srvutil.ServerUtil;
 import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
@@ -35,6 +38,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -45,7 +49,7 @@ public class RocketMQUtil {
     private static final List<String> existTopic = new ArrayList<>();
 
     //neither static topic nor compact topic. expansion with source topic.
-    public static void createNormalTopic(DefaultMQAdminExt mqAdmin, String topicName, int queueNum, Set<String> clusters) throws Exception {
+    public static void createNormalTopic(DefaultMQAdminExt mqAdmin, String topicName, int totalQueueNum, Set<String> clusters) throws Exception {
         if (check(mqAdmin, topicName)) {
             logger.info("topic[{}] already exist.", topicName);
             return;
@@ -55,20 +59,68 @@ public static void createNormalTopic(DefaultMQAdminExt mqAdmin, String topicName
             clusters = getCluster(mqAdmin);
         }
 
-        TopicConfig topicConfig = new TopicConfig(topicName, queueNum, queueNum, PermName.PERM_READ | PermName.PERM_WRITE);
 
         for (String cluster : clusters) {
             Set<String> masterSet = CommandUtil.fetchMasterAddrByClusterName(mqAdmin, cluster);
 
-            for (String addr : masterSet) {
-                mqAdmin.createAndUpdateTopicConfig(addr, topicConfig);
-                logger.info("create topic to broker:{} cluster:{}, success.", addr, cluster);
+            int remainder = totalQueueNum % masterSet.size();
+            if (remainder != 0) {
+                String temp = String.format("can not create topic:%s, total num=%s, master num=%s", topicName, totalQueueNum, masterSet.size());
+                logger.warn(temp);
             }
+
+            int queueNumInEachBroker = totalQueueNum / masterSet.size();
+            TopicConfig topicConfig = new TopicConfig(topicName, queueNumInEachBroker, queueNumInEachBroker, PermName.PERM_READ | PermName.PERM_WRITE);
+
+            if (remainder == 0) {
+                for (String addr : masterSet) {
+                    mqAdmin.createAndUpdateTopicConfig(addr, topicConfig);
+                    logger.info("create topic to broker:{} cluster:{}, success.", addr, cluster);
+                }
+            } else {
+                String[] masterArray = masterSet.toArray(new String[]{});
+
+                topicConfig = new TopicConfig(topicName, queueNumInEachBroker + remainder,
+                        queueNumInEachBroker + remainder, PermName.PERM_READ | PermName.PERM_WRITE);
+                mqAdmin.createAndUpdateTopicConfig(masterArray[0], topicConfig);
+
+                for (int i = 1; i < masterArray.length; i++) {
+                    topicConfig = new TopicConfig(topicName, queueNumInEachBroker, queueNumInEachBroker, PermName.PERM_READ | PermName.PERM_WRITE);
+                    mqAdmin.createAndUpdateTopicConfig(masterArray[0], topicConfig);
+                }
+            }
+
+        }
+    }
+
+    public static void createNormalTopic(DefaultMQAdminExt mqAdmin, String sourceTopic, String stateTopic) throws Exception {
+        //找到brokerAddr
+        TopicRouteData topicRouteData = mqAdmin.examineTopicRouteInfo(sourceTopic);
+        List<QueueData> queueData = topicRouteData.getQueueDatas();
+        List<BrokerData> brokerData = topicRouteData.getBrokerDatas();
+
+
+        HashMap<String, String> brokerName2MaterBrokerAddr = new HashMap<>();
+        for (BrokerData broker : brokerData) {
+            String masterBrokerAddr = broker.getBrokerAddrs().get(0L);
+            brokerName2MaterBrokerAddr.put(broker.getBrokerName(), masterBrokerAddr);
         }
+
+        for (QueueData queue : queueData) {
+            int readQueueNums = queue.getReadQueueNums();
+            int writeQueueNums = queue.getWriteQueueNums();
+            String brokerName = queue.getBrokerName();
+
+            TopicConfig topicConfig = new TopicConfig(stateTopic, readQueueNums, writeQueueNums);
+
+            mqAdmin.createAndUpdateTopicConfig(brokerName2MaterBrokerAddr.get(brokerName), topicConfig);
+        }
+
+        existTopic.add(stateTopic);
     }
 
     //used in RSQLDB,maybe.
-    public static void createStaticCompactTopic(DefaultMQAdminExt mqAdmin, String topicName, int queueNum, Set<String> clusters) throws Exception {
+    public static void createStaticCompactTopic(DefaultMQAdminExt mqAdmin, String topicName, int totalQueueNum, Set<String> clusters) throws Exception {
         if (check(mqAdmin, topicName)) {
             logger.info("topic[{}] already exist.", topicName);
             return;
@@ -80,15 +132,15 @@ public static void createStaticCompactTopic(DefaultMQAdminExt mqAdmin, String to
 
 
         for (String cluster : clusters) {
-            createStaticTopicWithCommand(topicName, queueNum, new HashSet<>(), cluster, mqAdmin.getNamesrvAddr());
-            logger.info("【step 1】create static topic:[{}] in cluster:[{}] success, logic queue num:[{}].", topicName, cluster, queueNum);
+            createStaticTopicWithCommand(topicName, totalQueueNum, new HashSet<>(), cluster, mqAdmin.getNamesrvAddr());
+            logger.info("【step 1】create static topic:[{}] in cluster:[{}] success, logic queue num:[{}].", topicName, cluster, totalQueueNum);
 
-            update2CompactTopicWithCommand(topicName, queueNum, cluster, mqAdmin.getNamesrvAddr());
+            update2CompactTopicWithCommand(topicName, totalQueueNum, cluster, mqAdmin.getNamesrvAddr());
             logger.info("【step 2】update static topic to compact topic success. topic:[{}], cluster:[{}]", topicName, cluster);
         }
 
         existTopic.add(topicName);
-        logger.info("create static-compact topic [{}] success, queue num [{}]", topicName, queueNum);
+        logger.info("create static-compact topic [{}] success, queue num [{}]", topicName, totalQueueNum);
     }
 
     public static void createStaticTopic(DefaultMQAdminExt mqAdmin, String topicName, int queueNum) throws Exception {
@@ -106,23 +158,23 @@ public static void createStaticTopic(DefaultMQAdminExt mqAdmin, String topicName
         existTopic.add(topicName);
     }
 
-    private static void createStaticTopicWithCommand(String topic, int queueNum, Set<String> brokers, String cluster, String nameservers) throws Exception {
+    private static void createStaticTopicWithCommand(String topic, int totalQueueNum, Set<String> brokers, String cluster, String nameservers) throws Exception {
         UpdateStaticTopicSubCommand cmd = new UpdateStaticTopicSubCommand();
         Options options = ServerUtil.buildCommandlineOptions(new Options());
         String[] args;
         if (cluster != null) {
             args = new String[]{
                     "-c", cluster,
                     "-t", topic,
-                    "-qn", String.valueOf(queueNum),
+                    "-qn", String.valueOf(totalQueueNum),
                     "-n", nameservers
             };
         } else {
             String brokerStr = String.join(",", brokers);
             args = new String[]{
                     "-b", brokerStr,
                     "-t", topic,
-                    "-qn", String.valueOf(queueNum),
+                    "-qn", String.valueOf(totalQueueNum),
                     "-n", nameservers
             };
         }