[ISSUE #9254] Optimize the logs of the message store

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

@@ -792,7 +792,12 @@ private boolean putMessagePositionInfo(final long offset, final int size, final
         final long cqOffset) {
 
         if (offset + size <= this.getMaxPhysicOffset()) {
-            log.warn("Maybe try to build consume queue repeatedly maxPhysicOffset={} phyOffset={}", this.getMaxPhysicOffset(), offset);
+            // During the recovery process after broker crashes, this logs will cause the scrolling of valid logs.
+            if (messageStore.getStateMachine().getCurrentState().isAfter(MessageStoreStateMachine.MessageStoreState.RECOVER_COMMITLOG_OK) ||
+                messageStore.getMessageStoreConfig().isEnableLogConsumeQueueRepeatedlyBuildWhenRecover()) {
+                log.warn("Maybe try to build consume queue repeatedly maxPhysicOffset={} phyOffset={}",
+                    this.getMaxPhysicOffset(), offset);
+            }
             return true;
         }
 

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

@@ -204,6 +204,8 @@ public class DefaultMessageStore implements MessageStore {
     // this is a unmodifiableMap
     private final ConcurrentMap<String, TopicConfig> topicConfigTable;
 
+    private final MessageStoreStateMachine stateMachine;
+
     private final ScheduledExecutorService scheduledCleanQueueExecutorService =
         ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("StoreCleanQueueScheduledThread"));
 
@@ -250,6 +252,8 @@ public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final Br
         lockFile = new RandomAccessFile(file, "rw");
 
         parseDelayLevel();
+
+        stateMachine = new MessageStoreStateMachine(LOGGER);
     }
 
     public ConsumeQueueStoreInterface createConsumeQueueStore() {
@@ -296,25 +300,28 @@ public boolean parseDelayLevel() {
     @Override
     public boolean load() {
         boolean result = true;
-
+        stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.LOAD_BEGIN);
         try {
             boolean lastExitOK = !this.isTempFileExist();
             LOGGER.info("last shutdown {}, store path root dir: {}",
                 lastExitOK ? "normally" : "abnormally", messageStoreConfig.getStorePathRootDir());
 
             // load Commit Log
             result = this.commitLog.load();
-
+            stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.LOAD_COMMITLOG_OK, result);
             // load Consume Queue
             result = result && this.consumeQueueStore.load();
+            stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.LOAD_CONSUME_QUEUE_OK, result);
 
             if (messageStoreConfig.isEnableCompaction()) {
                 result = result && this.compactionService.load(lastExitOK);
+                stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.LOAD_COMPACTION_OK, result);
             }
 
             if (result) {
                 loadCheckPoint();
                 result = this.indexService.load(lastExitOK);
+                stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.LOAD_INDEX_OK, result);
                 this.recover(lastExitOK);
                 LOGGER.info("message store recover end, and the max phy offset = {}", this.getMaxPhyOffset());
             }
@@ -343,28 +350,23 @@ public void loadCheckPoint() throws IOException {
     }
 
     private void recover(final boolean lastExitOK) throws RocksDBException {
+        this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.RECOVER_BEGIN);
         // recover consume queue
-        long recoverConsumeQueueStart = System.currentTimeMillis();
         this.consumeQueueStore.recover(this.brokerConfig.isRecoverConcurrently());
-        long dispatchFromPhyOffset = this.consumeQueueStore.getDispatchFromPhyOffset();
-        long recoverConsumeQueueEnd = System.currentTimeMillis();
+        this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.RECOVER_CONSUME_QUEUE_OK);
 
         // recover commitlog
+        long dispatchFromPhyOffset = this.consumeQueueStore.getDispatchFromPhyOffset();
         if (lastExitOK) {
             this.commitLog.recoverNormally(dispatchFromPhyOffset);
         } else {
             this.commitLog.recoverAbnormally(dispatchFromPhyOffset);
         }
+        this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.RECOVER_COMMITLOG_OK);
 
         // recover consume offset table
-        long recoverCommitLogEnd = System.currentTimeMillis();
         this.recoverTopicQueueTable();
-        long recoverConsumeOffsetEnd = System.currentTimeMillis();
-
-        LOGGER.info("message store recover total cost: {} ms, " +
-                "recoverConsumeQueue: {} ms, recoverCommitLog: {} ms, recoverOffsetTable: {} ms",
-            recoverConsumeOffsetEnd - recoverConsumeQueueStart, recoverConsumeQueueEnd - recoverConsumeQueueStart,
-            recoverCommitLogEnd - recoverConsumeQueueEnd, recoverConsumeOffsetEnd - recoverCommitLogEnd);
+        this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.RECOVER_TOPIC_QUEUE_TABLE_OK);
     }
 
     /**
@@ -411,6 +413,8 @@ public void start() throws Exception {
         this.addScheduleTask();
         this.perfs.start();
         this.shutdown = false;
+
+        this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.RUNNING);
     }
 
     private void doRecheckReputOffsetFromCq() throws InterruptedException {
@@ -470,6 +474,7 @@ private void doRecheckReputOffsetFromCq() throws InterruptedException {
     public void shutdown() {
         if (!this.shutdown) {
             this.shutdown = true;
+            this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.SHUTDOWN_BEGIN);
 
             this.scheduledExecutorService.shutdown();
             this.scheduledCleanQueueExecutorService.shutdown();
@@ -501,6 +506,7 @@ public void shutdown() {
             if (this.runningFlags.isWriteable() && dispatchBehindBytes() == 0) {
                 this.deleteFile(StorePathConfigHelper.getAbortFile(this.messageStoreConfig.getStorePathRootDir()));
                 shutDownNormal = true;
+                this.stateMachine.transitTo(MessageStoreStateMachine.MessageStoreState.SHUTDOWN_OK);
             } else {
                 LOGGER.warn("the store may be wrong, so shutdown abnormally, and keep abort file.");
             }
@@ -3001,4 +3007,8 @@ public ScheduledExecutorService getScheduledCleanQueueExecutorService() {
     public void destroyConsumeQueueStore(boolean loadAfterDestroy) {
         consumeQueueStore.destroy(loadAfterDestroy);
     }
+
+    public MessageStoreStateMachine getStateMachine() {
+        return stateMachine;
+    }
 }

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

@@ -983,4 +983,6 @@ DispatchRequest checkMessageAndReturnSize(final ByteBuffer byteBuffer, final boo
      * notify message arrive if necessary
      */
     void notifyMessageArriveIfNecessary(DispatchRequest dispatchRequest);
+
+    MessageStoreStateMachine getStateMachine();
 }

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

@@ -0,0 +1,120 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.store;
+
+import org.apache.rocketmq.common.constant.LoggerName;
+import org.apache.rocketmq.logging.org.slf4j.Logger;
+import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+
+public class MessageStoreStateMachine {
+    protected final Logger log;
+
+    private MessageStoreState currentState;
+    private long lastStateChangeTimestamp;
+    private final long startTimestamp;
+
+    public enum MessageStoreState {
+        INIT(0),
+
+        LOAD_BEGIN(10),
+        LOAD_COMMITLOG_OK(11),
+        LOAD_CONSUME_QUEUE_OK(12),
+        LOAD_COMPACTION_OK(13),
+        LOAD_INDEX_OK(14),
+
+        RECOVER_BEGIN(20),
+        RECOVER_CONSUME_QUEUE_OK(21),
+        RECOVER_COMMITLOG_OK(22),
+        RECOVER_TOPIC_QUEUE_TABLE_OK(23),
+
+        RUNNING(30),
+
+        SHUTDOWN_BEGIN(40),
+        SHUTDOWN_OK(41);
+
+        final int order;
+
+        MessageStoreState(int order) {
+            this.order = order;
+        }
+
+        public int getOrder() {
+            return order;
+        }
+
+        public boolean isBefore(MessageStoreState storeState) {
+            return this.order < storeState.order;
+        }
+
+        public boolean isAfter(MessageStoreState storeState) {
+            return this.order > storeState.order;
+        }
+    }
+
+
+    public MessageStoreStateMachine(Logger log) {
+        this.log = log == null ? LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME) : log;
+        this.currentState = MessageStoreState.INIT;
+        this.startTimestamp = System.currentTimeMillis();
+        this.lastStateChangeTimestamp = startTimestamp;
+        logStateChange(null, currentState, true);
+    }
+
+    public void transitTo(MessageStoreState newState) {
+        transitTo(newState, true);
+    }
+
+    public void transitTo(MessageStoreState newState, boolean success) {
+        if (!newState.isAfter(currentState)) {
+            throw new IllegalStateException(
+                String.format("Invalid state transition from %s to %s. Can only move forward.",
+                    currentState, newState)
+            );
+        }
+
+        logStateChange(currentState, newState, success);
+        if (success) {
+            this.currentState = newState;
+            this.lastStateChangeTimestamp = System.currentTimeMillis();
+        }
+    }
+
+    private void logStateChange(MessageStoreState fromState, MessageStoreState toState, boolean success) {
+        if (fromState == null && success) {
+            log.info("MessageStoreState initialized, state={}", toState);
+        } else if (success) {
+            log.info("MessageStoreState transition from {} to {}; Time in previous state={}ms, Total time={}ms",
+                fromState, toState, getCurrentStateRunningTimeMs(), getTotalRunningTimeMs());
+        } else {
+            log.warn("MessageStoreState transition from {} to {} failed; Time in previous state={}ms, Total "
+                + "time={}ms", fromState, toState, getCurrentStateRunningTimeMs(), getTotalRunningTimeMs());
+        }
+    }
+
+    public MessageStoreState getCurrentState() {
+        return currentState;
+    }
+
+    public long getTotalRunningTimeMs() {
+        return System.currentTimeMillis() - startTimestamp;
+    }
+
+    public long getCurrentStateRunningTimeMs() {
+        return System.currentTimeMillis() - lastStateChangeTimestamp;
+    }
+}

store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java

@@ -483,6 +483,8 @@ public void setRocksdbCompressionType(String compressionType) {
      **/
     private boolean useABSLock = false;
 
+    private boolean enableLogConsumeQueueRepeatedlyBuildWhenRecover = false;
+
     public boolean isRocksdbCQDoubleWriteEnable() {
         return rocksdbCQDoubleWriteEnable;
     }
@@ -2001,4 +2003,12 @@ public int getCombineCQMaxExtraSearchCommitLogFiles() {
     public void setCombineCQMaxExtraSearchCommitLogFiles(int combineCQMaxExtraSearchCommitLogFiles) {
         this.combineCQMaxExtraSearchCommitLogFiles = combineCQMaxExtraSearchCommitLogFiles;
     }
+
+    public boolean isEnableLogConsumeQueueRepeatedlyBuildWhenRecover() {
+        return enableLogConsumeQueueRepeatedlyBuildWhenRecover;
+    }
+
+    public void setEnableLogConsumeQueueRepeatedlyBuildWhenRecover(boolean enableLogConsumeQueueRepeatedlyBuildWhenRecover) {
+        this.enableLogConsumeQueueRepeatedlyBuildWhenRecover = enableLogConsumeQueueRepeatedlyBuildWhenRecover;
+    }
 }

store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java

@@ -42,6 +42,7 @@
 import org.apache.rocketmq.store.GetMessageResult;
 import org.apache.rocketmq.store.MessageFilter;
 import org.apache.rocketmq.store.MessageStore;
+import org.apache.rocketmq.store.MessageStoreStateMachine;
 import org.apache.rocketmq.store.PutMessageResult;
 import org.apache.rocketmq.store.QueryMessageResult;
 import org.apache.rocketmq.store.RunningFlags;
@@ -660,4 +661,9 @@ public void notifyMessageArriveIfNecessary(DispatchRequest dispatchRequest) {
     public MessageStore getNext() {
         return next;
     }
+
+    @Override
+    public MessageStoreStateMachine getStateMachine() {
+        return next.getStateMachine();
+    }
 }

store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java

@@ -17,6 +17,10 @@
 
 package org.apache.rocketmq.store;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
 import com.google.common.collect.Sets;
 import java.io.File;
 import java.io.RandomAccessFile;
@@ -35,21 +39,21 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import java.util.Random;
 import java.util.UUID;
-import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.message.MessageBatch;
 import org.apache.rocketmq.common.message.MessageDecoder;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageExtBatch;
 import org.apache.rocketmq.common.message.MessageExtBrokerInner;
-import org.apache.rocketmq.common.MixAll;
 import org.apache.rocketmq.store.config.BrokerRole;
 import org.apache.rocketmq.store.config.FlushDiskType;
 import org.apache.rocketmq.store.config.MessageStoreConfig;
@@ -66,10 +70,6 @@
 import org.junit.runner.RunWith;
 import org.mockito.junit.MockitoJUnitRunner;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
 @RunWith(MockitoJUnitRunner.class)
 public class DefaultMessageStoreTest {
     private final String storeMessage = "Once, there was a chance for me!";
@@ -911,7 +911,7 @@ public void testDeleteTopics() {
             String topicName = "topic-" + i;
             for (int j = 0; j < 4; j++) {
                 ConsumeQueue consumeQueue = new ConsumeQueue(topicName, j, messageStoreConfig.getStorePathRootDir(),
-                    messageStoreConfig.getMappedFileSizeConsumeQueue(), messageStore);
+                    messageStoreConfig.getMappedFileSizeConsumeQueue(), (DefaultMessageStore) messageStore);
                 cqTable.put(j, consumeQueue);
             }
             consumeQueueTable.put(topicName, cqTable);
@@ -933,7 +933,7 @@ public void testCleanUnusedTopic() {
             String topicName = "topic-" + i;
             for (int j = 0; j < 4; j++) {
                 ConsumeQueue consumeQueue = new ConsumeQueue(topicName, j, messageStoreConfig.getStorePathRootDir(),
-                    messageStoreConfig.getMappedFileSizeConsumeQueue(), messageStore);
+                    messageStoreConfig.getMappedFileSizeConsumeQueue(), (DefaultMessageStore) messageStore);
                 cqTable.put(j, consumeQueue);
             }
             consumeQueueTable.put(topicName, cqTable);