rebase onto develop

Change-Id: I193c1ba9a22d604729bbb7909477e0e4d7ad2a28

Author: f1amingo

broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConsumerOffsetManager.java

@@ -16,7 +16,7 @@
  */
 package org.apache.rocketmq.broker.config.v1;
 
-import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONWriter;
 import java.nio.file.Path;
 import java.nio.file.Paths;

broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java

@@ -183,7 +183,7 @@ public void deleteLmq(String parentTopic, String lmqName) {
                 String topicAtGroup = lmqName + TOPIC_GROUP_SEPARATOR + group;
                 brokerController.getConsumerOffsetManager().getOffsetTable().remove(topicAtGroup);
                 brokerController.getConsumerOffsetManager().removeConsumerOffset(topicAtGroup); // no iteration
-                brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager().getTable().remove(topicAtGroup);
+                brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager().remove(lmqName, group);
             });
             brokerController.getMessageStore().deleteTopics(Sets.newHashSet(lmqName));
             boolean sharding = brokerName.equals(liteSharding.shardingByLmqName(parentTopic, lmqName));

broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java

@@ -23,7 +23,7 @@
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
-import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
+import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.common.entity.ClientGroup;
@@ -498,7 +498,7 @@ public void onUnregister(String clientId, String group, String lmqName, boolean
                 LOGGER.info("unregister and reset offset. {}, {}, {}, {}", group, clientId, lmqName, consumerOffset);
                 if (consumerOffset > 0) {
                     consumerOffsetManager.assignResetOffset(lmqName, group, 0, 0);
-                    consumerOrderInfoManager.getTable().remove(lmqName + "@" + group); // may cause race condition.
+                    consumerOrderInfoManager.remove(lmqName, group);
                 }
             }
         }

broker/src/main/java/org/apache/rocketmq/broker/offset/MemoryConsumerOrderInfoManager.java

@@ -18,6 +18,7 @@
 package org.apache.rocketmq.broker.offset;
 
 import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.pop.orderly.QueueLevelConsumerManager;
 
 /**
  * Memory-based Consumer Order Information Manager for Lite Topics
@@ -30,7 +31,7 @@
  * <p>
  * We may make structural adjustments and optimizations to reduce overhead and memory footprint.
  */
-public class MemoryConsumerOrderInfoManager extends ConsumerOrderInfoManager {
+public class MemoryConsumerOrderInfoManager extends QueueLevelConsumerManager {
 
     public MemoryConsumerOrderInfoManager(BrokerController brokerController) {
         super(brokerController);

broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManager.java

@@ -68,6 +68,20 @@ void update(String attemptId, boolean isRetry, String topic, String group, int q
      */
     boolean checkBlock(String attemptId, String topic, String group, int queueId, long invisibleTime);
 
+    /**
+     * Remove the specified topic and group
+     * Usually called during topic deletion
+     *
+     * @param topic Topic name
+     * @param group Consumer group name
+     */
+    void remove(String topic, String group);
+
+    /**
+     * Get order info count
+     */
+    int getOrderInfoCount();
+
     /**
      * Commit message and calculate next consumption offset
      * Called when consumer ACKs messages
@@ -137,6 +151,7 @@ void updateNextVisibleTime(String topic, String group, int queueId, long queueOf
      * Get available message result
      * Used to retrieve messages from cache
      */
-    CompletableFuture<GetMessageResult> getAvailableMessageResult(String attemptId, long popTime, long invisibleTime, String groupId,
+    CompletableFuture<GetMessageResult> getAvailableMessageResult(String attemptId, long popTime, long invisibleTime,
+        String groupId,
         String topicId, int queueId, int batchSize, StringBuilder orderCountInfoBuilder);
 }

broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java

@@ -179,6 +179,16 @@ public void clearBlock(String topic, String group, int queueId) {
         });
     }
 
+    @Override
+    public void remove(String topic, String group) {
+        table.remove(buildKey(topic, group));
+    }
+
+    @Override
+    public int getOrderInfoCount() {
+        return table.size();
+    }
+
     @Override
     public OrderedConsumptionLevel getOrderedConsumptionLevel() {
         return OrderedConsumptionLevel.QUEUE;
@@ -383,7 +393,7 @@ public CompletableFuture<GetMessageResult> getAvailableMessageResult(String atte
     }
 
     @VisibleForTesting
-    QueueLevelConsumerOrderInfoLockManager getConsumerOrderInfoLockManager() {
+    protected QueueLevelConsumerOrderInfoLockManager getConsumerOrderInfoLockManager() {
         return queueLevelConsumerOrderInfoLockManager;
     }
 

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

@@ -105,7 +105,7 @@ protected RemotingCommand getBrokerLiteInfo(ChannelHandlerContext ctx,
         body.setMaxLmqNum(brokerController.getMessageStoreConfig().getMaxLmqConsumeQueueNum());
         body.setCurrentLmqNum(brokerController.getMessageStore().getQueueStore().getLmqNum());
         body.setLiteSubscriptionCount(brokerController.getLiteSubscriptionRegistry().getActiveSubscriptionNum());
-        body.setOrderInfoCount(brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager().getTable().size());
+        body.setOrderInfoCount(brokerController.getPopLiteMessageProcessor().getConsumerOrderInfoManager().getOrderInfoCount());
         body.setCqTableSize(brokerController.getMessageStore().getQueueStore().getConsumeQueueTable().size());
         body.setOffsetTableSize(brokerController.getConsumerOffsetManager().getOffsetTable().size());
         body.setEventMapSize(brokerController.getLiteEventDispatcher().getEventMapSize());

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

@@ -25,9 +25,9 @@
 import org.apache.rocketmq.broker.longpolling.PollingResult;
 import org.apache.rocketmq.broker.longpolling.PopLiteLongPollingService;
 import org.apache.rocketmq.broker.metrics.LiteConsumerLagCalculator;
-import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
 import org.apache.rocketmq.broker.offset.MemoryConsumerOrderInfoManager;
 import org.apache.rocketmq.broker.pop.PopConsumerLockService;
+import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
 import org.apache.rocketmq.common.KeyBuilder;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.Pair;
@@ -341,7 +341,7 @@ public Pair<StringBuilder, GetMessageResult> handleGetMessageResult(GetMessageRe
         StringBuilder orderCountInfo = new StringBuilder();
         if (GetMessageStatus.FOUND.equals(result.getStatus()) && !result.getMessageQueueOffset().isEmpty()) {
             consumerOrderInfoManager.update(attemptId, false, lmqName, group, 0,
-                popTime, invisibleTime, result.getMessageQueueOffset(), orderCountInfo);
+                popTime, invisibleTime, result.getMessageQueueOffset(), orderCountInfo, null);
             recordPopLiteMetrics(result, parentTopic, group);
             orderCountInfo = transformOrderCountInfo(orderCountInfo, result.getMessageCount());
         }

broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java

@@ -24,7 +24,7 @@
 
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.config.v1.RocksDBConsumerOffsetManager;
-import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
+import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
 import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.broker.topic.TopicConfigManager;
@@ -86,8 +86,6 @@ public class AbstractLiteLifecycleManagerTest {
     private final TopicConfig topicConfig = new TopicConfig(PARENT_TOPIC, 1, 1);
     private final SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
     private final ConcurrentMap<String, ConcurrentMap<Integer, Long>> offsetTable = new ConcurrentHashMap<>();
-    private final ConcurrentHashMap<String, ConcurrentHashMap<Integer, ConsumerOrderInfoManager.OrderInfo>>
-        orderInfoTable = new ConcurrentHashMap<>();
 
     @Before
     public void setUp() {
@@ -115,7 +113,7 @@ public void setUp() {
         when(subscriptionGroupManager.getSubscriptionGroupTable()).thenReturn(groupTable);
 
         when(consumerOffsetManager.getOffsetTable()).thenReturn(offsetTable);
-        when(consumerOrderInfoManager.getTable()).thenReturn(orderInfoTable);
+        when(consumerOffsetManager.getPullOffsetTable()).thenReturn(offsetTable);
 
         TestLiteLifecycleManager testObject = new TestLiteLifecycleManager(brokerController, liteSharding);
         lifecycleManager = Mockito.spy(testObject);
@@ -127,7 +125,6 @@ public void reset() {
         topicConfig.getAttributes().clear();
         groupConfig.getAttributes().clear();
         offsetTable.clear();
-        orderInfoTable.clear();
     }
 
     @Test
@@ -201,29 +198,24 @@ public void testDeleteLmq() {
         String removeKey = EXIST_LMQ_NAME + TOPIC_GROUP_SEPARATOR + GROUP;
         offsetTable.put(otherKey, new ConcurrentHashMap<>());
         offsetTable.put(removeKey, new ConcurrentHashMap<>());
-        orderInfoTable.put(otherKey, new ConcurrentHashMap<>());
-        orderInfoTable.put(removeKey, new ConcurrentHashMap<>());
 
         // sharding to this broker
         when(liteSharding.shardingByLmqName(PARENT_TOPIC, EXIST_LMQ_NAME)).thenReturn(brokerConfig.getBrokerName());
         lifecycleManager.deleteLmq(PARENT_TOPIC, EXIST_LMQ_NAME);
 
         Assert.assertTrue(offsetTable.containsKey(otherKey));
         Assert.assertFalse(offsetTable.containsKey(removeKey));
-        Assert.assertTrue(orderInfoTable.containsKey(otherKey));
-        Assert.assertFalse(orderInfoTable.containsKey(removeKey));
         verify(consumerOffsetManager).removeConsumerOffset(removeKey);
         verify(messageStore).deleteTopics(Collections.singleton(EXIST_LMQ_NAME));
         verify(liteSubscriptionRegistry).cleanSubscription(EXIST_LMQ_NAME, false);
+        verify(consumerOrderInfoManager, times(1)).remove(EXIST_LMQ_NAME, GROUP);
 
         // not sharding to this broker
         when(liteSharding.shardingByLmqName(PARENT_TOPIC, EXIST_LMQ_NAME)).thenReturn("otherBrokerName");
         lifecycleManager.deleteLmq(PARENT_TOPIC, EXIST_LMQ_NAME);
 
         Assert.assertTrue(offsetTable.containsKey(otherKey));
         Assert.assertFalse(offsetTable.containsKey(removeKey));
-        Assert.assertTrue(orderInfoTable.containsKey(otherKey));
-        Assert.assertFalse(orderInfoTable.containsKey(removeKey));
         verify(consumerOffsetManager, times(2)).removeConsumerOffset(removeKey);
         verify(messageStore, times(2)).deleteTopics(Collections.singleton(EXIST_LMQ_NAME));
         verify(liteSubscriptionRegistry, times(2)).cleanSubscription(EXIST_LMQ_NAME, false);

broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java

@@ -22,7 +22,7 @@
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.longpolling.PopLiteLongPollingService;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
-import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
+import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
 import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.common.BrokerConfig;