diff --git a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java b/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java index 79e5368216b..2850299b7d6 100644 --- a/store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java +++ b/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; } diff --git a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java index ea0a814caf2..99eaa4b43c9 100644 --- a/store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java +++ b/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 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,7 +300,7 @@ 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: {}", @@ -304,17 +308,20 @@ public boolean load() { // 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; + } } diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java b/store/src/main/java/org/apache/rocketmq/store/MessageStore.java index 9c9a556f6d3..52c2de33fd3 100644 --- a/store/src/main/java/org/apache/rocketmq/store/MessageStore.java +++ b/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(); } diff --git a/store/src/main/java/org/apache/rocketmq/store/MessageStoreStateMachine.java b/store/src/main/java/org/apache/rocketmq/store/MessageStoreStateMachine.java new file mode 100644 index 00000000000..e5e07676dc7 --- /dev/null +++ b/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; + } +} diff --git a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java b/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java index 28ab74eb353..60f6a90381c 100644 --- a/store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java +++ b/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; + } } diff --git a/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java index 74c67d01621..19ace1c8e3a 100644 --- a/store/src/main/java/org/apache/rocketmq/store/plugin/AbstractPluginMessageStore.java +++ b/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(); + } } diff --git a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java b/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java index eee38e0a8f4..ac25ac5430b 100644 --- a/store/src/test/java/org/apache/rocketmq/store/DefaultMessageStoreTest.java +++ b/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); diff --git a/store/src/test/java/org/apache/rocketmq/store/MessageStoreStateMachineTest.java b/store/src/test/java/org/apache/rocketmq/store/MessageStoreStateMachineTest.java new file mode 100644 index 00000000000..b6f424147d2 --- /dev/null +++ b/store/src/test/java/org/apache/rocketmq/store/MessageStoreStateMachineTest.java @@ -0,0 +1,147 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.anyLong; +import static org.mockito.Mockito.verify; + +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.store.MessageStoreStateMachine.MessageStoreState; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class MessageStoreStateMachineTest { + + private Logger mockLogger; + private MessageStoreStateMachine stateMachine; + + @BeforeEach + void setUp() { + // Mock Logger + mockLogger = Mockito.mock(Logger.class); + + // Initialize StateMachine + stateMachine = new MessageStoreStateMachine(mockLogger); + } + + /** + * Test the constructor of MessageStoreStateMachine. + */ + @Test + void testConstructor() { + // Verify initial state + assertEquals(MessageStoreState.INIT, stateMachine.getCurrentState()); + + // Verify logger was called for initialization + verify(mockLogger).info(anyString(), eq(MessageStoreState.INIT)); + } + + /** + * Test valid state transition in transitTo method. + */ + @Test + void testValidStateTransition() { + // Perform a valid state transition + stateMachine.transitTo(MessageStoreState.LOAD_COMMITLOG_OK); + + // Verify the current state is updated + assertEquals(MessageStoreState.LOAD_COMMITLOG_OK, stateMachine.getCurrentState()); + + // Verify logger was called for state transition + verify(mockLogger).info(anyString(), eq(MessageStoreState.INIT), eq(MessageStoreState.LOAD_COMMITLOG_OK), + anyLong(), anyLong()); + } + + /** + * Test fail state transition in transitTo method. + */ + @Test + void testValidFailStateTransition() { + stateMachine.transitTo(MessageStoreState.LOAD_COMMITLOG_OK, false); + assertEquals(MessageStoreState.INIT, stateMachine.getCurrentState()); + verify(mockLogger).warn(anyString(), eq(MessageStoreState.INIT), eq(MessageStoreState.LOAD_COMMITLOG_OK), + anyLong(), anyLong()); + } + + /** + * Test invalid state transition in transitTo method. + */ + @Test + void testInvalidStateTransition() { + // Perform an invalid state transition + Exception exception = assertThrows(IllegalStateException.class, () -> { + stateMachine.transitTo(MessageStoreState.INIT); + }); + + // Verify the exception message + String expectedMessage = "Invalid state transition from INIT to INIT. Can only move forward."; + assertEquals(expectedMessage, exception.getMessage()); + } + + /** + * Test getCurrentState method. + */ + @Test + void testGetCurrentState() { + // Verify the current state + assertEquals(MessageStoreState.INIT, stateMachine.getCurrentState()); + } + + /** + * Test getTotalRunningTimeMs method. + */ + @Test + void testGetTotalRunningTimeMs() { + // Sleep for a short duration to simulate elapsed time + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // Verify the total running time is approximately correct + long totalTime = stateMachine.getTotalRunningTimeMs(); + assertTrue(totalTime >= 100 && totalTime < 200); + } + + /** + * Test getCurrentStateRunningTimeMs method. + */ + @Test + void testGetCurrentStateRunningTimeMs() { + // Perform a state transition + stateMachine.transitTo(MessageStoreState.LOAD_COMMITLOG_OK); + + // Sleep for a short duration to simulate elapsed time + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + // Verify the current state running time is approximately correct + long currentStateTime = stateMachine.getCurrentStateRunningTimeMs(); + assertTrue(currentStateTime >= 100 && currentStateTime < 200); + } +}