fix unit test

Change-Id: Ic811d4e0af800f145fb94122985965e1b0569915

Author: f1amingo

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

@@ -197,8 +197,7 @@ public void commitOffset(String clientHost, String group, String topic, int queu
             }
         }
         if (versionChangeCounter.incrementAndGet() % brokerController.getBrokerConfig().getConsumerOffsetUpdateVersionStep() == 0) {
-            long stateMachineVersion = brokerController.getMessageStore() != null ? brokerController.getMessageStore().getStateMachineVersion() : 0;
-            dataVersion.nextVersion(stateMachineVersion);
+            updateDataVersion();
         }
         if (!brokerController.getBrokerConfig().isPersistConsumerOffsetIncrementally()) {
             return;

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

@@ -515,7 +515,7 @@ protected RemotingCommand ackLite(AckMessageRequestHeader requestHeader, BatchAc
         String lmqName = LiteUtil.toLmqName(requestHeader.getTopic(), requestHeader.getLiteTopic());
         long ackOffset = requestHeader.getOffset();
         long maxOffset = this.brokerController.getLiteLifecycleManager().getMaxOffsetInQueue(lmqName);
-        if (ackOffset > maxOffset) { // TODO moling. minOffset
+        if (ackOffset > maxOffset) {
             POP_LOGGER.warn("ack lite offset illegal, {}, {}, {}", lmqName, ackOffset, maxOffset);
             response.setCode(ResponseCode.NO_MESSAGE);
             response.setRemark("ack offset illegal.");

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

@@ -372,7 +372,7 @@ protected CompletableFuture<RemotingCommand> processChangeInvisibleTimeForLite(
         }
         String lmqName = LiteUtil.toLmqName(requestHeader.getTopic(), requestHeader.getLiteTopic());
         long maxOffset = this.brokerController.getLiteLifecycleManager().getMaxOffsetInQueue(lmqName);
-        if (requestHeader.getOffset() > maxOffset) { // TODO moling. minOffset
+        if (requestHeader.getOffset() > maxOffset) {
             POP_LOGGER.warn("process lite offset illegal, {}, {}, {}", lmqName, requestHeader.getOffset(), maxOffset);
             response.setCode(ResponseCode.NO_MESSAGE);
             return CompletableFuture.completedFuture(response);

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

@@ -372,7 +372,8 @@ public StringBuilder transformOrderCountInfo(StringBuilder orderCountInfo, int m
         }
     }
 
-    private void recordPopLiteMetrics(GetMessageResult result, String parentTopic, String group) {
+    @VisibleForTesting
+    protected void recordPopLiteMetrics(GetMessageResult result, String parentTopic, String group) {
         Attributes attributes = this.brokerController.getBrokerMetricsManager().newAttributesBuilder()
             .put(LABEL_TOPIC, parentTopic)
             .put(LABEL_CONSUMER_GROUP, group)

broker/src/test/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculatorTest.java

@@ -285,9 +285,9 @@ public void testGetLagCountTopK_NormalCase() {
         LiteConsumerLagCalculator spyCalculator = spy(liteConsumerLagCalculator);
 
         // Mock getStoreTimestamp method on the spy
-        doReturn(timestamp1).when(spyCalculator).getStoreTimestamp(lmqName1, consumerOffset1 + 1);
-        doReturn(timestamp2).when(spyCalculator).getStoreTimestamp(lmqName2, consumerOffset2 + 1);
-        doReturn(timestamp3).when(spyCalculator).getStoreTimestamp(lmqName3, consumerOffset3 + 1);
+        doReturn(timestamp1).when(spyCalculator).getStoreTimestamp(lmqName1, consumerOffset1);
+        doReturn(timestamp2).when(spyCalculator).getStoreTimestamp(lmqName2, consumerOffset2);
+        doReturn(timestamp3).when(spyCalculator).getStoreTimestamp(lmqName3, consumerOffset3);
 
         // Mock getMaxOffset method on the spy
         doReturn(100L).when(spyCalculator).getMaxOffset(lmqName1);

broker/src/test/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessorTest.java

@@ -731,7 +731,7 @@ public void testSearchOffsetByTimestampWithLiteTopic() throws Exception {
 
         SearchOffsetRequestHeader requestHeader = new SearchOffsetRequestHeader();
         requestHeader.setTopic(topic);
-        requestHeader.setQueueId(1);
+        requestHeader.setQueueId(0);
         requestHeader.setTimestamp(timestamp);
         requestHeader.setLiteTopic(liteTopic);
         requestHeader.setBoundaryType(BoundaryType.LOWER);

broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java

@@ -33,6 +33,7 @@
 import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
 import org.apache.rocketmq.broker.metrics.LiteConsumerLagCalculator;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
 import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
 import org.apache.rocketmq.broker.topic.TopicConfigManager;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -123,6 +124,9 @@ public class LiteManagerProcessorTest {
     @Mock
     private LiteEventDispatcher liteEventDispatcher;
 
+    @Mock
+    private PopLiteMessageProcessor popLiteMessageProcessor;
+
     private LiteManagerProcessor processor;
 
     @Before
@@ -136,9 +140,17 @@ public void setUp() {
         when(brokerController.getLiteSubscriptionRegistry()).thenReturn(liteSubscriptionRegistry);
         when(brokerController.getBrokerMetricsManager()).thenReturn(brokerMetricsManager);
         when(brokerController.getLiteEventDispatcher()).thenReturn(liteEventDispatcher);
+        when(brokerController.getPopLiteMessageProcessor()).thenReturn(popLiteMessageProcessor);
+        when(brokerController.getConsumerOffsetManager()).thenReturn(consumerOffsetManager);
+
+        ConsumerOrderInfoManager consumerOrderInfoManager = new ConsumerOrderInfoManager(brokerController);
+        when(popLiteMessageProcessor.getConsumerOrderInfoManager()).thenReturn(consumerOrderInfoManager);
 
         when(messageStore.getQueueStore()).thenReturn(consumeQueueStore);
+        when(consumeQueueStore.getConsumeQueueTable()).thenReturn(new ConcurrentHashMap<>());
         when(brokerMetricsManager.getLiteConsumerLagCalculator()).thenReturn(liteConsumerLagCalculator);
+
+        when(consumerOffsetManager.getOffsetTable()).thenReturn(new ConcurrentHashMap<>());
     }
 
     @Test
@@ -700,6 +712,11 @@ public void testTriggerLiteDispatch() throws Exception {
         RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.TRIGGER_LITE_DISPATCH, requestHeader);
         request.makeCustomHeaderToNet();
 
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        groupConfig.setGroupName(group);
+        groupConfig.setLiteBindTopic("parent_topic");
+        when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig);
+
         RemotingCommand response = processor.triggerLiteDispatch(ctx, request);
 
         assertNotNull(response);

broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java

@@ -59,9 +59,11 @@
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.times;
@@ -336,6 +338,8 @@ public void testHandleGetMessageResult_found() {
         getResult.getMessageQueueOffset().add(0L);
         getResult.getMessageQueueOffset().add(1L);
 
+        doNothing().when(popLiteMessageProcessor).recordPopLiteMetrics(any(), anyString(), anyString());
+
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.handleGetMessageResult(
             getResult, "parentTopic", "group", "lmqName", System.currentTimeMillis(), 6000L, "attemptId");
 

common/src/main/java/org/apache/rocketmq/common/TopicAttributes.java

@@ -18,6 +18,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import org.apache.rocketmq.common.attribute.Attribute;
 import org.apache.rocketmq.common.attribute.EnumAttribute;
 import org.apache.rocketmq.common.attribute.LongRangeAttribute;
@@ -56,7 +57,7 @@ public class TopicAttributes {
         "lite.topic.expiration",
         true,
         -1,
-        Long.MAX_VALUE,
+        TimeUnit.DAYS.toMinutes(30),
         -1
     );