fix

Author: qianye1001

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

@@ -3550,7 +3550,7 @@ private boolean processConsumeQueuesForTopic(ConcurrentMap<Integer, ConsumeQueue
 
             long minOffsetByTime = 0L;
             try {
-                minOffsetByTime = rocksDBMessageStore.getConsumeQueueStore().getOffsetInQueueByTime(topic, queueId, checkpointByStoreTime, BoundaryType.UPPER);
+                minOffsetByTime = rocksDBMessageStore.getQueueStore().getOffsetInQueueByTime(topic, queueId, checkpointByStoreTime, BoundaryType.UPPER);
             } catch (Exception e) {
                 // ignore
             }

broker/src/main/java/org/apache/rocketmq/broker/schedule/ScheduleMessageService.java

@@ -236,7 +236,7 @@ public boolean correctDelayOffset() {
         try {
             for (int delayLevel : delayLevelTable.keySet()) {
                 ConsumeQueueInterface cq =
-                    brokerController.getMessageStore().getQueueStore().findOrCreateConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
+                    brokerController.getMessageStore().findConsumeQueue(TopicValidator.RMQ_SYS_SCHEDULE_TOPIC,
                         delayLevel2QueueId(delayLevel));
                 Long currentDelayOffset = offsetTable.get(delayLevel);
                 if (currentDelayOffset == null || cq == null) {

broker/src/test/java/org/apache/rocketmq/broker/offset/RocksdbTransferOffsetAndCqTest.java

@@ -135,7 +135,7 @@ public void testRocksdbCqWrite() throws RocksDBException {
             return;
         }
         RocksDBMessageStore kvStore = defaultMessageStore.getRocksDBMessageStore();
-        ConsumeQueueStoreInterface store = kvStore.getConsumeQueueStore();
+        ConsumeQueueStoreInterface store = kvStore.getQueueStore();
         store.start();
         ConsumeQueueInterface rocksdbCq = defaultMessageStore.getRocksDBMessageStore().findConsumeQueue(topic, queueId);
         ConsumeQueueInterface fileCq = defaultMessageStore.findConsumeQueue(topic, queueId);

client/src/test/java/org/apache/rocketmq/client/impl/consumer/ProcessQueueTest.java

@@ -16,6 +16,11 @@
  */
 package org.apache.rocketmq.client.impl.consumer;
 
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.TreeMap;
 import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.exception.MQBrokerException;
@@ -29,12 +34,6 @@
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import java.lang.reflect.Field;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.TreeMap;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -158,7 +157,6 @@ public void testProcessQueue() {
         ProcessQueue processQueue2 = createProcessQueue();
         assertEquals(processQueue1.getMsgAccCnt(), processQueue2.getMsgAccCnt());
         assertEquals(processQueue1.getTryUnlockTimes(), processQueue2.getTryUnlockTimes());
-        assertEquals(processQueue1.getLastLockTimestamp(), processQueue2.getLastLockTimestamp());
         assertEquals(processQueue1.getLastPullTimestamp(), processQueue2.getLastPullTimestamp());
     }
 

store/src/main/java/org/apache/rocketmq/store/CommitLog.java

@@ -17,6 +17,8 @@
 package org.apache.rocketmq.store;
 
 import com.google.common.base.Strings;
+import com.sun.jna.NativeLong;
+import com.sun.jna.Pointer;
 import java.net.Inet6Address;
 import java.net.InetSocketAddress;
 import java.nio.ByteBuffer;
@@ -34,8 +36,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import com.sun.jna.NativeLong;
-import com.sun.jna.Pointer;
 import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.SystemClock;
@@ -66,7 +66,6 @@
 import org.apache.rocketmq.store.logfile.MappedFile;
 import org.apache.rocketmq.store.util.LibC;
 import org.rocksdb.RocksDBException;
-
 import sun.nio.ch.DirectBuffer;
 
 /**
@@ -408,8 +407,7 @@ else if (!dispatchRequest.isSuccess()) {
             log.warn("The commitlog files are deleted, and delete the consume queue files");
             this.mappedFileQueue.setFlushedWhere(0);
             this.mappedFileQueue.setCommittedWhere(0);
-            this.defaultMessageStore.getQueueStore().destroy();
-            this.defaultMessageStore.getQueueStore().loadAfterDestroy();
+            this.defaultMessageStore.destroyConsumeQueueStore(true);
         }
     }
 
@@ -818,8 +816,7 @@ else if (size == 0) {
             log.warn("The commitlog files are deleted, and delete the consume queue files");
             this.mappedFileQueue.setFlushedWhere(0);
             this.mappedFileQueue.setCommittedWhere(0);
-            this.defaultMessageStore.getQueueStore().destroy();
-            this.defaultMessageStore.getQueueStore().loadAfterDestroy();
+            this.defaultMessageStore.destroyConsumeQueueStore(true);
         }
     }
 

store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java

@@ -35,6 +35,7 @@
 import org.apache.rocketmq.store.config.StorePathConfigHelper;
 import org.apache.rocketmq.store.logfile.MappedFile;
 import org.apache.rocketmq.store.queue.ConsumeQueueInterface;
+import org.apache.rocketmq.store.queue.ConsumeQueueStoreInterface;
 import org.apache.rocketmq.store.queue.CqUnit;
 import org.apache.rocketmq.store.queue.FileQueueLifeCycle;
 import org.apache.rocketmq.store.queue.MultiDispatchUtils;
@@ -61,6 +62,7 @@ public class ConsumeQueue implements ConsumeQueueInterface, FileQueueLifeCycle {
     private static final Logger LOG_ERROR = LoggerFactory.getLogger(LoggerName.STORE_ERROR_LOGGER_NAME);
 
     private final MessageStore messageStore;
+    private final ConsumeQueueStoreInterface consumeQueueStore;
 
     private final MappedFileQueue mappedFileQueue;
     private final String topic;
@@ -83,9 +85,20 @@ public ConsumeQueue(
         final String storePath,
         final int mappedFileSize,
         final MessageStore messageStore) {
+        this(topic, queueId, storePath, mappedFileSize, messageStore, messageStore.getQueueStore());
+    }
+
+    public ConsumeQueue(
+        final String topic,
+        final int queueId,
+        final String storePath,
+        final int mappedFileSize,
+        final MessageStore messageStore,
+        final ConsumeQueueStoreInterface consumeQueueStore) {
         this.storePath = storePath;
         this.mappedFileSize = mappedFileSize;
         this.messageStore = messageStore;
+        this.consumeQueueStore = consumeQueueStore;
 
         this.topic = topic;
         this.queueId = queueId;
@@ -899,14 +912,14 @@ public CqUnit get(long offset) {
     @Override
     public Pair<CqUnit, Long> getCqUnitAndStoreTime(long index) {
         CqUnit cqUnit = get(index);
-        Long messageStoreTime = this.messageStore.getQueueStore().getStoreTime(cqUnit);
+        Long messageStoreTime = this.consumeQueueStore.getStoreTime(cqUnit);
         return new Pair<>(cqUnit, messageStoreTime);
     }
 
     @Override
     public Pair<CqUnit, Long> getEarliestUnitAndStoreTime() {
         CqUnit cqUnit = getEarliestUnit();
-        Long messageStoreTime = this.messageStore.getQueueStore().getStoreTime(cqUnit);
+        Long messageStoreTime = this.consumeQueueStore.getStoreTime(cqUnit);
         return new Pair<>(cqUnit, messageStoreTime);
     }