[ISSUE #4758] Timer Message adapts to slaveActingMaster mode (#4757)

* Timer Message adapts to slaveActingMaster mode * Add asf-header to TimerCheckPointTest
2026-08-30 18:10:44 +08:00 · 2022-08-04 10:11:02 +08:00
parent 838854ffcd
commit c292574cd5
9 changed files with 502 additions and 48 deletions
@@ -732,6 +732,7 @@ public class BrokerController {
                    this.timerCheckpoint = new TimerCheckpoint(BrokerPathConfigHelper.getTimerCheckPath(messageStoreConfig.getStorePathRootDir()));
                    TimerMetrics timerMetrics = new TimerMetrics(BrokerPathConfigHelper.getTimerMetricsPath(messageStoreConfig.getStorePathRootDir()));
                    this.timerMessageStore = new TimerMessageStore(messageStore, messageStoreConfig, timerCheckpoint, timerMetrics, brokerStatsManager);
+                    this.timerMessageStore.registerEscapeBridgeHook(msg -> escapeBridge.putMessage(msg));
                    this.messageStore.setTimerMessageStore(this.timerMessageStore);
                }
            } catch (IOException e) {
@@ -1962,6 +1963,10 @@ public class BrokerController {
                this.scheduleMessageService.stop();
            }
            isScheduleServiceStart = shouldStart;
+
+            if (timerMessageStore != null) {
+                timerMessageStore.setShouldRunningDequeue(shouldStart);
+            }
        }
    }

@@ -39,6 +39,7 @@ import org.apache.rocketmq.store.config.StorePathConfigHelper;
 import org.apache.rocketmq.broker.schedule.DelayOffsetSerializeWrapper;
 import org.apache.rocketmq.store.ha.HAConnectionState;
 import org.apache.rocketmq.store.ha.HAConnectionStateNotificationRequest;
+import org.apache.rocketmq.store.timer.TimerCheckpoint;

 public class BrokerPreOnlineService extends ServiceThread {
    private static final InternalLogger LOGGER = InternalLoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
@@ -159,6 +160,8 @@ public class BrokerPreOnlineService extends ServiceThread {

            ConsumerOffsetSerializeWrapper consumerOffsetSerializeWrapper = this.brokerController.getBrokerOuterAPI().getAllConsumerOffset(brokerAddr);

+            TimerCheckpoint timerCheckpoint = this.brokerController.getBrokerOuterAPI().getTimerCheckPoint(brokerAddr);
+
            if (null != consumerOffsetSerializeWrapper && brokerController.getConsumerOffsetManager().getDataVersion().compare(consumerOffsetSerializeWrapper.getDataVersion()) <= 0) {
                LOGGER.info("{}'s consumerOffset data version is larger than master broker, {}'s consumerOffset will be used.", brokerAddr, brokerAddr);
                this.brokerController.getConsumerOffsetManager().getOffsetTable()
@@ -180,6 +183,14 @@ public class BrokerPreOnlineService extends ServiceThread {
                }
            }

+            if (null != this.brokerController.getTimerCheckpoint() && this.brokerController.getTimerCheckpoint().getDataVersion().compare(timerCheckpoint.getDataVersion()) <= 0) {
+                LOGGER.info("{}'s timerCheckpoint data version is larger than master broker, {}'s timerCheckpoint will be used.", brokerAddr, brokerAddr);
+                this.brokerController.getTimerCheckpoint().setLastReadTimeMs(timerCheckpoint.getLastReadTimeMs());
+                this.brokerController.getTimerCheckpoint().setMasterTimerQueueOffset(timerCheckpoint.getMasterTimerQueueOffset());
+                this.brokerController.getTimerCheckpoint().getDataVersion().assignNewOne(timerCheckpoint.getDataVersion());
+                this.brokerController.getTimerCheckpoint().flush();
+            }
+
            for (BrokerAttachedPlugin brokerAttachedPlugin : brokerController.getBrokerAttachedPlugins()) {
                if (brokerAttachedPlugin != null) {
                    brokerAttachedPlugin.syncMetadataReverse(brokerAddr);
@@ -16,6 +16,8 @@
 */
 package org.apache.rocketmq.store.timer;

+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.rocketmq.common.DataVersion;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.constant.LoggerName;
 import org.apache.rocketmq.logging.InternalLogger;
@@ -38,6 +40,7 @@ public class TimerCheckpoint {
    private volatile long lastTimerLogFlushPos = 0;
    private volatile long lastTimerQueueOffset = 0;
    private volatile long masterTimerQueueOffset = 0; // read from master
+    private final DataVersion dataVersion = new DataVersion();

    public TimerCheckpoint() {
        this.randomAccessFile = null;
@@ -60,12 +63,21 @@ public class TimerCheckpoint {
            this.lastTimerLogFlushPos = this.mappedByteBuffer.getLong(8);
            this.lastTimerQueueOffset = this.mappedByteBuffer.getLong(16);
            this.masterTimerQueueOffset = this.mappedByteBuffer.getLong(24);
+            // new add to record dataVersion
+            if (this.mappedByteBuffer.hasRemaining()) {
+                dataVersion.setStateVersion(this.mappedByteBuffer.getLong(32));
+                dataVersion.setTimestamp(this.mappedByteBuffer.getLong(40));
+                dataVersion.setCounter(new AtomicLong(this.mappedByteBuffer.getLong(48)));
+            }

            log.info("timer checkpoint file lastReadTimeMs " + this.lastReadTimeMs + ", "
                + UtilAll.timeMillisToHumanString(this.lastReadTimeMs));
            log.info("timer checkpoint file lastTimerLogFlushPos " + this.lastTimerLogFlushPos);
            log.info("timer checkpoint file lastTimerQueueOffset " + this.lastTimerQueueOffset);
            log.info("timer checkpoint file masterTimerQueueOffset " + this.masterTimerQueueOffset);
+            log.info("timer checkpoint file data version state version " + this.dataVersion.getStateVersion());
+            log.info("timer checkpoint file data version timestamp " + this.dataVersion.getTimestamp());
+            log.info("timer checkpoint file data version counter " + this.dataVersion.getCounter());
        } else {
            log.info("timer checkpoint file not exists, " + scpPath);
        }
@@ -96,6 +108,10 @@ public class TimerCheckpoint {
        this.mappedByteBuffer.putLong(8, this.lastTimerLogFlushPos);
        this.mappedByteBuffer.putLong(16, this.lastTimerQueueOffset);
        this.mappedByteBuffer.putLong(24, this.masterTimerQueueOffset);
+        // new add to record dataVersion
+        this.mappedByteBuffer.putLong(32, this.dataVersion.getStateVersion());
+        this.mappedByteBuffer.putLong(40, this.dataVersion.getTimestamp());
+        this.mappedByteBuffer.putLong(48, this.dataVersion.getCounter().get());
        this.mappedByteBuffer.force();
    }

@@ -104,11 +120,15 @@ public class TimerCheckpoint {
    }

    public static ByteBuffer encode(TimerCheckpoint another) {
-        ByteBuffer byteBuffer = ByteBuffer.allocate(32);
+        ByteBuffer byteBuffer = ByteBuffer.allocate(56);
        byteBuffer.putLong(another.getLastReadTimeMs());
        byteBuffer.putLong(another.getLastTimerLogFlushPos());
        byteBuffer.putLong(another.getLastTimerQueueOffset());
        byteBuffer.putLong(another.getMasterTimerQueueOffset());
+        // new add to record dataVersion
+        byteBuffer.putLong(another.getDataVersion().getStateVersion());
+        byteBuffer.putLong(another.getDataVersion().getTimestamp());
+        byteBuffer.putLong(another.getDataVersion().getCounter().get());
        byteBuffer.flip();
        return byteBuffer;
    }
@@ -119,6 +139,12 @@ public class TimerCheckpoint {
        tmp.setLastTimerLogFlushPos(byteBuffer.getLong());
        tmp.setLastTimerQueueOffset(byteBuffer.getLong());
        tmp.setMasterTimerQueueOffset(byteBuffer.getLong());
+        // new add to record dataVersion
+        if (byteBuffer.hasRemaining()) {
+            tmp.getDataVersion().setStateVersion(byteBuffer.getLong());
+            tmp.getDataVersion().setTimestamp(byteBuffer.getLong());
+            tmp.getDataVersion().setCounter(new AtomicLong(byteBuffer.getLong()));
+        }
        return tmp;
    }

@@ -149,4 +175,12 @@ public class TimerCheckpoint {
    public void setMasterTimerQueueOffset(final long masterTimerQueueOffset) {
        this.masterTimerQueueOffset = masterTimerQueueOffset;
    }
+
+    public void updateDateVersion(long stateVersion) {
+        dataVersion.nextVersion(stateVersion);
+    }
+
+    public DataVersion getDataVersion() {
+        return dataVersion;
+    }
 }
@@ -17,6 +17,7 @@
 package org.apache.rocketmq.store.timer;

 import com.conversantmedia.util.concurrent.DisruptorBlockingQueue;
+import java.util.function.Function;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.rocketmq.common.ServiceThread;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
@@ -118,6 +119,8 @@ public class TimerMessageStore {
    private volatile long commitReadTimeMs;
    private volatile long currQueueOffset; //only one queue that is 0
    private volatile long commitQueueOffset;
+    private volatile long lastCommitReadTimeMs;
+    private volatile long lastCommitQueueOffset;

    private long lastEnqueueButExpiredTime;
    private long lastEnqueueButExpiredStoreTime;
@@ -136,10 +139,14 @@ public class TimerMessageStore {
    private boolean dequeueStatusChangeFlag = false;
    private long shouldStartTime;

+    // True if current store is master or current brokerId is equal to the minimum brokerId of the replica group in slaveActingMaster mode.
+    private volatile boolean shouldRunningDequeue;
    private final BrokerStatsManager brokerStatsManager;
+    private Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook;

    public TimerMessageStore(final MessageStore messageStore, final MessageStoreConfig storeConfig,
-                             TimerCheckpoint timerCheckpoint, TimerMetrics timerMetrics, final BrokerStatsManager brokerStatsManager) throws IOException {
+        TimerCheckpoint timerCheckpoint, TimerMetrics timerMetrics,
+        final BrokerStatsManager brokerStatsManager) throws IOException {
        this.messageStore = messageStore;
        this.storeConfig = storeConfig;
        this.commitLogFileSize = storeConfig.getMappedFileSizeCommitLog();
@@ -148,7 +155,7 @@ public class TimerMessageStore {
        // TimerWheel contains the fixed number of slots regardless of precision.
        this.slotsTotal = TIMER_WHELL_TTL_DAY * DAY_SECS;
        this.timerWheel = new TimerWheel(getTimerWheelPath(storeConfig.getStorePathRootDir()),
-                this.slotsTotal, precisionMs);
+            this.slotsTotal, precisionMs);
        this.timerLog = new TimerLog(getTimerLogPath(storeConfig.getStorePathRootDir()), timerLogFileSize);
        this.timerMetrics = timerMetrics;
        this.timerCheckpoint = timerCheckpoint;
@@ -186,7 +193,9 @@ public class TimerMessageStore {
            dequeueGetMessageServices[i] = new TimerDequeueGetMessageService();
        }
        int putThreadNum = storeConfig.getTimerPutMessageThreadNum();
-        if (putThreadNum <= 0) putThreadNum = 1;
+        if (putThreadNum <= 0) {
+            putThreadNum = 1;
+        }
        dequeuePutMessageServices = new TimerDequeuePutMessageService[putThreadNum];
        for (int i = 0; i < dequeuePutMessageServices.length; i++) {
            dequeuePutMessageServices[i] = new TimerDequeuePutMessageService();
@@ -200,7 +209,6 @@ public class TimerMessageStore {
            dequeueGetQueue = new LinkedBlockingDeque<List<TimerRequest>>(1024);
            dequeuePutQueue = new LinkedBlockingDeque<TimerRequest>(1024);
        }
-
        this.brokerStatsManager = brokerStatsManager;
    }

@@ -280,11 +288,12 @@ public class TimerMessageStore {
            recoverAndRevise(minFirst, false);
        }
        LOGGER.info("Timer recover ok currReadTimerMs:{} currQueueOffset:{} checkQueueOffset:{} processOffset:{}",
-                currReadTimeMs, currQueueOffset, timerCheckpoint.getLastTimerQueueOffset(), processOffset);
+            currReadTimeMs, currQueueOffset, timerCheckpoint.getLastTimerQueueOffset(), processOffset);

        commitReadTimeMs = currReadTimeMs;
        commitQueueOffset = currQueueOffset;

+        prepareTimerCheckPoint();
    }

    public long reviseQueueOffset(long processOffset) {
@@ -314,7 +323,7 @@ public class TimerMessageStore {
            while (maxCount-- > 0) {
                if (tmpOffset < 0) {
                    LOGGER.warn("reviseQueueOffset check cq offset fail, msg in cq is not found.{}, {}",
-                            offsetPy, sizePy);
+                        offsetPy, sizePy);
                    break;
                }
                SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(tmpOffset);
@@ -328,7 +337,7 @@ public class TimerMessageStore {
                    int sizePyTemp = bufferCQ.getByteBuffer().getInt();
                    if (offsetPyTemp == offsetPy && sizePyTemp == sizePy) {
                        LOGGER.info("reviseQueueOffset check cq offset ok. {}, {}, {}",
-                                tmpOffset, offsetPyTemp, sizePyTemp);
+                            tmpOffset, offsetPyTemp, sizePyTemp);
                        cqOffset = tmpOffset;
                        break;
                    }
@@ -349,7 +358,7 @@ public class TimerMessageStore {
    //recover timerlog and revise timerwheel
    //return process offset
    private long recoverAndRevise(long beginOffset, boolean checkTimerLog) {
-        LOGGER.info("Begin fto recover timerlog offset:{} check:{}", beginOffset, checkTimerLog);
+        LOGGER.info("Begin to recover timerlog offset:{} check:{}", beginOffset, checkTimerLog);
        MappedFile lastFile = timerLog.getMappedFileQueue().getLastMappedFile();
        if (null == lastFile) {
            return 0;
@@ -472,6 +481,11 @@ public class TimerMessageStore {
        LOGGER.info("Timer start ok currReadTimerMs:[{}] queueOffset:[{}]", new Timestamp(currReadTimeMs), currQueueOffset);
    }

+    public void start(boolean shouldRunningDequeue) {
+        this.shouldRunningDequeue = shouldRunningDequeue;
+        this.start();
+    }
+
    public void shutdown() {
        if (SHUTDOWN == state) {
            return;
@@ -541,7 +555,7 @@ public class TimerMessageStore {

    private boolean isRunningEnqueue() {
        checkBrokerRole();
-        if (!isMaster() && currQueueOffset >= timerCheckpoint.getMasterTimerQueueOffset()) {
+        if (!shouldRunningDequeue && !isMaster() && currQueueOffset >= timerCheckpoint.getMasterTimerQueueOffset()) {
            return false;
        }

@@ -549,14 +563,22 @@ public class TimerMessageStore {
    }

    private boolean isRunningDequeue() {
-        if (!isMaster()) {
-            currReadTimeMs = timerCheckpoint.getLastReadTimeMs();
-            commitReadTimeMs = currReadTimeMs;
+        if (!this.shouldRunningDequeue) {
+            syncLastReadTimeMs();
            return false;
        }
        return isRunning();
    }

+    public void syncLastReadTimeMs() {
+        currReadTimeMs = timerCheckpoint.getLastReadTimeMs();
+        commitReadTimeMs = currReadTimeMs;
+    }
+
+    public void setShouldRunningDequeue(final boolean shouldRunningDequeue) {
+        this.shouldRunningDequeue = shouldRunningDequeue;
+    }
+
    public void addMetric(MessageExt msg, int value) {
        try {
            if (null == msg || null == msg.getProperty(MessageConst.PROPERTY_REAL_TOPIC)) {
@@ -694,7 +716,7 @@ public class TimerMessageStore {
            // If it's a delete message, then slot's total num -1
            // TODO: check if the delete msg is in the same slot with "the msg to be deleted".
            timerWheel.putSlot(delayedTime, slot.firstPos == -1 ? ret : slot.firstPos, ret,
-                    isDelete ? slot.num - 1 : slot.num + 1, slot.magic);
+                isDelete ? slot.num - 1 : slot.num + 1, slot.magic);
            addMetric(messageExt, isDelete ? -1 : 1);
        }
        return -1 != ret;
@@ -807,8 +829,8 @@ public class TimerMessageStore {
        int checkNum = 0;
        while (true) {
            if (dequeuePutQueue.size() > 0
-                    || !checkStateForGetMessages(AbstractStateService.WAITING)
-                    || !checkStateForPutMessages(AbstractStateService.WAITING)) {
+                || !checkStateForGetMessages(AbstractStateService.WAITING)
+                || !checkStateForPutMessages(AbstractStateService.WAITING)) {
                //let it go
            } else {
                checkNum++;
@@ -1006,15 +1028,19 @@ public class TimerMessageStore {

    //0 succ; 1 fail, need retry; 2 fail, do not retry;
    private int doPut(MessageExtBrokerInner message, boolean roll) throws Exception {
-        if (lastBrokerRole == BrokerRole.SLAVE) {
-            LOGGER.warn("Trying do put timer msg in slave, [{}]", message);
-            return PUT_NO_RETRY;
-        }
+
        if (!roll && null != message.getProperty(MessageConst.PROPERTY_TIMER_DEL_UNIQKEY)) {
            LOGGER.warn("Trying do put delete timer msg:[{}] roll:[{}]", message, roll);
            return PUT_NO_RETRY;
        }
-        PutMessageResult putMessageResult = messageStore.putMessage(message);
+
+        PutMessageResult putMessageResult = null;
+        if (escapeBridgeHook != null) {
+            putMessageResult = escapeBridgeHook.apply(message);
+        } else {
+            putMessageResult = messageStore.putMessage(message);
+        }
+
        int retryNum = 0;
        while (retryNum < 3) {
            if (null == putMessageResult || null == putMessageResult.getPutMessageStatus()) {
@@ -1025,7 +1051,7 @@ public class TimerMessageStore {
                        if (brokerStatsManager != null) {
                            this.brokerStatsManager.incTopicPutNums(message.getTopic(), 1, 1);
                            this.brokerStatsManager.incTopicPutSize(message.getTopic(),
-                                    putMessageResult.getAppendMessageResult().getWroteBytes());
+                                putMessageResult.getAppendMessageResult().getWroteBytes());
                            this.brokerStatsManager.incBrokerPutNums(1);
                        }
                        return PUT_OK;
@@ -1058,7 +1084,7 @@ public class TimerMessageStore {
        MessageAccessor.setProperties(msgInner, msgExt.getProperties());
        TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
        long tagsCodeValue =
-                MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
+            MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
        msgInner.setTagsCode(tagsCodeValue);
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));

@@ -1101,8 +1127,8 @@ public class TimerMessageStore {
                int hash = hashTopicForMetrics(entry.getKey());
                if (smallHashs.containsKey(hash)) {
                    LOGGER.warn("[CheckAndReviseMetrics]Metric hash collision between small-small code:{} small topic:{}{} small topic:{}{}", hash,
-                            entry.getKey(), entry.getValue(),
-                            smallHashs.get(hash), smallOnes.get(smallHashs.get(hash)));
+                        entry.getKey(), entry.getValue(),
+                        smallHashs.get(hash), smallOnes.get(smallHashs.get(hash)));
                    smallHashCollisions.add(hash);
                }
                smallHashs.put(hash, entry.getKey());
@@ -1118,8 +1144,8 @@ public class TimerMessageStore {
                    Map.Entry<String, TimerMetrics.Metric> smallEntry = smalllIt.next();
                    if (hashTopicForMetrics(smallEntry.getKey()) == hashTopicForMetrics(bjgEntry.getKey())) {
                        LOGGER.warn("[CheckAndReviseMetrics]Metric hash collision between small-big code:{} small topic:{}{} big topic:{}{}", hashTopicForMetrics(smallEntry.getKey()),
-                                smallEntry.getKey(), smallEntry.getValue(),
-                                bjgEntry.getKey(), bjgEntry.getValue());
+                            smallEntry.getKey(), smallEntry.getValue(),
+                            bjgEntry.getKey(), bjgEntry.getValue());
                        smalllIt.remove();
                    }
                }
@@ -1276,7 +1302,7 @@ public class TimerMessageStore {
                            req.setLatch(latch);
                            try {
                                perfs.startTick("enqueue_put");
-                                if (isMaster() && req.getDelayTime() < currWriteTimeMs) {
+                                if (shouldRunningDequeue && req.getDelayTime() < currWriteTimeMs) {
                                    dequeuePutQueue.put(req);
                                } else {
                                    boolean doEnqueueRes = doEnqueue(req.getOffsetPy(), req.getSizePy(), req.getDelayTime(), req.getMsg());
@@ -1574,11 +1600,11 @@ public class TimerMessageStore {
                        ConsumeQueue cq = (ConsumeQueue) messageStore.getConsumeQueue(TIMER_TOPIC, 0);
                        long maxOffsetInQueue = cq == null ? 0 : cq.getMaxOffsetInQueue();
                        TimerMessageStore.LOGGER.info("[{}]Timer progress-check commitRead:[{}] currRead:[{}] currWrite:[{}] readBehind:{} currReadOffset:{} offsetBehind:{} behindMaster:{} " +
-                                        "enqPutQueue:{} deqGetQueue:{} deqPutQueue:{} allCongestNum:{} enqExpiredStoreTime:{}",
-                                storeConfig.getBrokerRole(),
-                                format(commitReadTimeMs), format(currReadTimeMs), format(currWriteTimeMs), getReadBehind(),
-                                tmpQueueOffset, maxOffsetInQueue - tmpQueueOffset, timerCheckpoint.getMasterTimerQueueOffset() - tmpQueueOffset,
-                                enqueuePutQueue.size(), dequeueGetQueue.size(), dequeuePutQueue.size(), getALlCongestNum(), format(lastEnqueueButExpiredStoreTime));
+                                "enqPutQueue:{} deqGetQueue:{} deqPutQueue:{} allCongestNum:{} enqExpiredStoreTime:{}",
+                            storeConfig.getBrokerRole(),
+                            format(commitReadTimeMs), format(currReadTimeMs), format(currWriteTimeMs), getReadBehind(),
+                            tmpQueueOffset, maxOffsetInQueue - tmpQueueOffset, timerCheckpoint.getMasterTimerQueueOffset() - tmpQueueOffset,
+                            enqueuePutQueue.size(), dequeueGetQueue.size(), dequeuePutQueue.size(), getALlCongestNum(), format(lastEnqueueButExpiredStoreTime));
                    }
                    timerMetrics.persist();
                    waitForRunning(storeConfig.getTimerFlushIntervalMs());
@@ -1640,13 +1666,22 @@ public class TimerMessageStore {

    public void prepareTimerCheckPoint() {
        timerCheckpoint.setLastTimerLogFlushPos(timerLog.getMappedFileQueue().getFlushedWhere());
-        if (isMaster()) {
-            timerCheckpoint.setLastReadTimeMs(commitReadTimeMs);
+        timerCheckpoint.setLastReadTimeMs(commitReadTimeMs);
+        if (shouldRunningDequeue) {
            timerCheckpoint.setMasterTimerQueueOffset(commitQueueOffset);
+            if (commitReadTimeMs != lastCommitReadTimeMs || commitQueueOffset != lastCommitQueueOffset) {
+                timerCheckpoint.updateDateVersion(messageStore.getStateMachineVersion());
+                lastCommitReadTimeMs = commitReadTimeMs;
+                lastCommitQueueOffset = commitQueueOffset;
+            }
        }
        timerCheckpoint.setLastTimerQueueOffset(Math.min(commitQueueOffset, timerCheckpoint.getMasterTimerQueueOffset()));
    }

+    public void registerEscapeBridgeHook(Function<MessageExtBrokerInner, PutMessageResult> escapeBridgeHook) {
+        this.escapeBridgeHook = escapeBridgeHook;
+    }
+
    public boolean isMaster() {
        return BrokerRole.SLAVE != lastBrokerRole;
    }
@@ -0,0 +1,127 @@
+/*
+ * 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.timer;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class TimerCheckPointTest {
+
+    private String baseDir;
+
+    @Before
+    public void init() throws IOException {
+        baseDir = StoreTestUtils.createBaseDir();
+    }
+
+    @Test
+    public void testCheckPoint() throws IOException {
+        String baseSrc = baseDir + File.separator + "timercheck";
+        TimerCheckpoint first = new TimerCheckpoint(baseSrc);
+        assertEquals(0, first.getLastReadTimeMs());
+        assertEquals(0, first.getLastTimerLogFlushPos());
+        assertEquals(0, first.getLastTimerQueueOffset());
+        assertEquals(0, first.getMasterTimerQueueOffset());
+        first.setLastReadTimeMs(1000);
+        first.setLastTimerLogFlushPos(1100);
+        first.setLastTimerQueueOffset(1200);
+        first.setMasterTimerQueueOffset(1300);
+        first.shutdown();
+        TimerCheckpoint second = new TimerCheckpoint(baseSrc);
+        assertEquals(1000, second.getLastReadTimeMs());
+        assertEquals(1100, second.getLastTimerLogFlushPos());
+        assertEquals(1200, second.getLastTimerQueueOffset());
+        assertEquals(1300, second.getMasterTimerQueueOffset());
+    }
+
+    @Test
+    public void testNewCheckPoint() throws IOException {
+        String baseSrc = baseDir + File.separator + "timercheck2";
+        TimerCheckpoint first = new TimerCheckpoint(baseSrc);
+        assertEquals(0, first.getLastReadTimeMs());
+        assertEquals(0, first.getLastTimerLogFlushPos());
+        assertEquals(0, first.getLastTimerQueueOffset());
+        assertEquals(0, first.getMasterTimerQueueOffset());
+        assertEquals(0, first.getDataVersion().getStateVersion());
+        assertEquals(0, first.getDataVersion().getCounter().get());
+        first.setLastReadTimeMs(1000);
+        first.setLastTimerLogFlushPos(1100);
+        first.setLastTimerQueueOffset(1200);
+        first.setMasterTimerQueueOffset(1300);
+        first.getDataVersion().setStateVersion(1400);
+        first.getDataVersion().setTimestamp(1500);
+        first.getDataVersion().setCounter(new AtomicLong(1600));
+        first.shutdown();
+        TimerCheckpoint second = new TimerCheckpoint(baseSrc);
+        assertEquals(1000, second.getLastReadTimeMs());
+        assertEquals(1100, second.getLastTimerLogFlushPos());
+        assertEquals(1200, second.getLastTimerQueueOffset());
+        assertEquals(1300, second.getMasterTimerQueueOffset());
+        assertEquals(1400, second.getDataVersion().getStateVersion());
+        assertEquals(1500, second.getDataVersion().getTimestamp());
+        assertEquals(1600, second.getDataVersion().getCounter().get());
+    }
+
+    @Test
+    public void testEncodeDecode() throws IOException {
+        TimerCheckpoint first = new TimerCheckpoint();
+        first.setLastReadTimeMs(1000);
+        first.setLastTimerLogFlushPos(1100);
+        first.setLastTimerQueueOffset(1200);
+        first.setMasterTimerQueueOffset(1300);
+
+        TimerCheckpoint second = TimerCheckpoint.decode(TimerCheckpoint.encode(first));
+        assertEquals(first.getLastReadTimeMs(), second.getLastReadTimeMs());
+        assertEquals(first.getLastTimerLogFlushPos(), second.getLastTimerLogFlushPos());
+        assertEquals(first.getLastTimerQueueOffset(), second.getLastTimerQueueOffset());
+        assertEquals(first.getMasterTimerQueueOffset(), second.getMasterTimerQueueOffset());
+    }
+
+    @Test
+    public void testNewEncodeDecode() throws IOException {
+        TimerCheckpoint first = new TimerCheckpoint();
+        first.setLastReadTimeMs(1000);
+        first.setLastTimerLogFlushPos(1100);
+        first.setLastTimerQueueOffset(1200);
+        first.setMasterTimerQueueOffset(1300);
+        first.getDataVersion().setStateVersion(1400);
+        first.getDataVersion().setTimestamp(1500);
+        first.getDataVersion().setCounter(new AtomicLong(1600));
+        TimerCheckpoint second = TimerCheckpoint.decode(TimerCheckpoint.encode(first));
+        assertEquals(first.getLastReadTimeMs(), second.getLastReadTimeMs());
+        assertEquals(first.getLastTimerLogFlushPos(), second.getLastTimerLogFlushPos());
+        assertEquals(first.getLastTimerQueueOffset(), second.getLastTimerQueueOffset());
+        assertEquals(first.getMasterTimerQueueOffset(), second.getMasterTimerQueueOffset());
+        assertEquals(first.getDataVersion().getStateVersion(), 1400);
+        assertEquals(first.getDataVersion().getTimestamp(), 1500);
+        assertEquals(first.getDataVersion().getCounter().get(), 1600);
+    }
+
+    @After
+    public void shutdown() {
+        if (null != baseDir) {
+            StoreTestUtils.deleteFile(baseDir);
+        }
+    }
+}
@@ -64,8 +64,6 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;

-//// Timer unit tests are very unstable, ignore these temporarily
-//@Ignore
 public class TimerMessageStoreTest {
    private final byte[] msgBody = new byte[1024];
    private static MessageStore messageStore;
@@ -173,7 +171,7 @@ public class TimerMessageStoreTest {

        final TimerMessageStore timerMessageStore = createTimerMessageStore(null);
        timerMessageStore.load();
-        timerMessageStore.start();
+        timerMessageStore.start(true);

        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
        long delayMs = curr + 3000;
@@ -220,7 +218,7 @@ public class TimerMessageStoreTest {
        storeConfig.setTimerCongestNumEachSlot(100);
        TimerMessageStore timerMessageStore = createTimerMessageStore(null);
        timerMessageStore.load();
-        timerMessageStore.start();
+        timerMessageStore.start(true);

        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
        // Make sure delayMs won't be over.
@@ -265,7 +263,7 @@ public class TimerMessageStoreTest {

        TimerMessageStore timerMessageStore = createTimerMessageStore(null);
        timerMessageStore.load();
-        timerMessageStore.start();
+        timerMessageStore.start(true);

        long delayMs = System.currentTimeMillis() - 2 * precisionMs;
        for (int i = 0; i < 10; i++) {
@@ -289,7 +287,7 @@ public class TimerMessageStoreTest {

        TimerMessageStore timerMessageStore = createTimerMessageStore(null);
        timerMessageStore.load();
-        timerMessageStore.start();
+        timerMessageStore.start(true);

        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
        long delayMs = curr + 1000;
@@ -326,7 +324,7 @@ public class TimerMessageStoreTest {

        final TimerMessageStore timerMessageStore = createTimerMessageStore(null);
        timerMessageStore.load();
-        timerMessageStore.start();
+        timerMessageStore.start(true);

        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
        final long delayMs = curr + 1000;
@@ -373,7 +371,7 @@ public class TimerMessageStoreTest {
        String base = StoreTestUtils.createBaseDir();
        final TimerMessageStore first = createTimerMessageStore(base);
        first.load();
-        first.start();
+        first.start(true);

        final int msgNum = 250;
        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
@@ -423,7 +421,7 @@ public class TimerMessageStoreTest {
        assertEquals(second.getCommitQueueOffset(), second.getQueueOffset());
        assertEquals(second.getCurrReadTimeMs(), second.getCommitReadTimeMs());
        assertEquals(first.getCommitReadTimeMs(), second.getCommitReadTimeMs());
-        second.start();
+        second.start(true);

        // Wait until all messages have wrote back to commitLog and consumeQueue.
        await().atMost(5000, TimeUnit.MILLISECONDS).until(new Callable<Boolean>() {
@@ -447,7 +445,7 @@ public class TimerMessageStoreTest {

        TimerMessageStore first = createTimerMessageStore(null);
        first.load();
-        first.start();
+        first.start(true);

        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
        long delaySec = storeConfig.getTimerMaxDelaySec() + 20;
@@ -469,7 +467,7 @@ public class TimerMessageStoreTest {

        TimerMessageStore timerMessageStore = createTimerMessageStore(null);
        timerMessageStore.load();
-        timerMessageStore.start();
+        timerMessageStore.start(true);

        long curr = System.currentTimeMillis() / precisionMs * precisionMs;
        long delayMs = curr + 4 * precisionMs;
@@ -228,4 +228,76 @@ public class GetMetadataReverseIT extends ContainerIntegrationTestBase {

        pushConsumer.shutdown();
    }
+
+    @Test
+    public void testGetMetadataReverse_timerCheckPoint() throws Exception {
+        String topic = GetMetadataReverseIT.class.getSimpleName() + "_timerCheckPoint" + random.nextInt(65535);
+        createTopicTo(master1With3Replicas, topic, 1, 1);
+        createTopicTo(master2With3Replicas, topic, 1, 1);
+        createTopicTo(master3With3Replicas, topic, 1, 1);
+        // Wait topic synchronization
+        await().atMost(Duration.ofMinutes(1)).until(() -> {
+            InnerSalveBrokerController slaveBroker = brokerContainer2.getSlaveBrokers().iterator().next();
+            return slaveBroker.getTopicConfigManager().selectTopicConfig(topic) != null;
+        });
+
+        DefaultMQPushConsumer pushConsumer = createPushConsumer(CONSUMER_GROUP);
+        pushConsumer.subscribe(topic, "*");
+        AtomicInteger receivedMsgCount = new AtomicInteger(0);
+        pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+            receivedMsgCount.addAndGet(msgs.size());
+            msgs.forEach(x -> System.out.printf(x + "%n"));
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+        pushConsumer.start();
+
+        MessageQueue messageQueue = new MessageQueue(topic, master1With3Replicas.getBrokerConfig().getBrokerName(), 0);
+        int sendSuccess = 0;
+        for (int i = 0; i < MESSAGE_COUNT; i++) {
+            Message msg = new Message(topic, Integer.toString(i).getBytes());
+            msg.setDelayTimeSec(30);
+            SendResult sendResult = producer.send(msg, messageQueue);
+            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
+                sendSuccess++;
+            }
+        }
+        final int finalSendSuccess = sendSuccess;
+        await().atMost(Duration.ofMinutes(1)).until(() -> finalSendSuccess >= MESSAGE_COUNT);
+        System.out.printf("send success%n");
+
+        isolateBroker(master1With3Replicas);
+        brokerContainer1.removeBroker(new BrokerIdentity(
+            master1With3Replicas.getBrokerConfig().getBrokerClusterName(),
+            master1With3Replicas.getBrokerConfig().getBrokerName(),
+            master1With3Replicas.getBrokerConfig().getBrokerId()));
+
+        System.out.printf("Remove master%n");
+
+        await().atMost(Duration.ofMinutes(1)).until(() -> receivedMsgCount.get() >= MESSAGE_COUNT);
+
+        await().atMost(Duration.ofMinutes(1)).until(() -> {
+            pushConsumer.getDefaultMQPushConsumerImpl().persistConsumerOffset();
+            Map<Integer, Long> OffsetTable = master2With3Replicas.getConsumerOffsetManager().queryOffset(CONSUMER_GROUP, topic);
+            if (OffsetTable != null) {
+                long totalOffset = 0;
+                for (final Long offset : OffsetTable.values()) {
+                    totalOffset += offset;
+                }
+                return totalOffset >= MESSAGE_COUNT;
+            }
+            return false;
+        });
+
+        //Add back master
+        master1With3Replicas = brokerContainer1.addBroker(master1With3Replicas.getBrokerConfig(), master1With3Replicas.getMessageStoreConfig());
+        master1With3Replicas.start();
+        cancelIsolatedBroker(master1With3Replicas);
+        System.out.printf("Add back master%n");
+
+        awaitUntilSlaveOK();
+
+        await().atMost(Duration.ofMinutes(1)).until(() -> master1With3Replicas.getTimerCheckpoint().getMasterTimerQueueOffset() >= MESSAGE_COUNT);
+
+        pushConsumer.shutdown();
+    }
 }
@@ -39,7 +39,7 @@ import org.junit.Test;

 import static org.awaitility.Awaitility.await;

-//The test is correct, but it takes too much time, so it is ignored for the time being
+//The test is correct, but it takes too much time and not core functions, so it is ignored for the time being
@Ignore
 public class ScheduleSlaveActingMasterIT extends ContainerIntegrationTestBase {

@@ -135,6 +135,65 @@ public class ScheduleSlaveActingMasterIT extends ContainerIntegrationTestBase {
        Thread.sleep(30000);
    }

+    @Test
+    public void testLocalActing_timerMsg() throws Exception {
+        awaitUntilSlaveOK();
+        String topic = ScheduleSlaveActingMasterIT.class.getSimpleName() + random.nextInt(65535);
+        createTopic(topic);
+        DefaultMQPushConsumer pushConsumer = createPushConsumer(CONSUME_GROUP);
+        pushConsumer.subscribe(topic, "*");
+        AtomicInteger receivedMsgCount = new AtomicInteger(0);
+        AtomicInteger inTimeMsgCount = new AtomicInteger(0);
+        pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+            long period = System.currentTimeMillis() - msgs.get(0).getBornTimestamp();
+            if (Math.abs(period - 30000) <= 1000) {
+                inTimeMsgCount.addAndGet(msgs.size());
+            }
+            receivedMsgCount.addAndGet(msgs.size());
+            msgs.forEach(x -> System.out.printf(x + "%n"));
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+        pushConsumer.start();
+
+        MessageQueue messageQueue = new MessageQueue(topic, master1With3Replicas.getBrokerConfig().getBrokerName(), 0);
+        int sendSuccess = 0;
+        for (int i = 0; i < MESSAGE_COUNT; i++) {
+            Message msg = new Message(topic, MESSAGE_BODY);
+            msg.setDelayTimeSec(30);
+            SendResult sendResult = producer.send(msg, messageQueue);
+            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
+                sendSuccess++;
+            }
+        }
+        final int finalSendSuccess = sendSuccess;
+        await().atMost(Duration.ofMinutes(2)).until(() -> finalSendSuccess >= MESSAGE_COUNT);
+        System.out.printf("send success%n");
+
+        isolateBroker(master1With3Replicas);
+        brokerContainer1.removeBroker(new BrokerIdentity(
+            master1With3Replicas.getBrokerConfig().getBrokerClusterName(),
+            master1With3Replicas.getBrokerConfig().getBrokerName(),
+            master1With3Replicas.getBrokerConfig().getBrokerId()));
+
+        System.out.printf("Remove master1%n");
+
+        await().atMost(Duration.ofMinutes(1)).until(() -> receivedMsgCount.get() >= MESSAGE_COUNT && inTimeMsgCount.get() >= MESSAGE_COUNT * 0.95);
+
+        System.out.printf("consumer received %d msg, %d in time%n", receivedMsgCount.get(), inTimeMsgCount.get());
+
+        pushConsumer.shutdown();
+
+        //Add back master
+        master1With3Replicas = brokerContainer1.addBroker(master1With3Replicas.getBrokerConfig(), master1With3Replicas.getMessageStoreConfig());
+        master1With3Replicas.start();
+        cancelIsolatedBroker(master1With3Replicas);
+        System.out.printf("Add back master1%n");
+
+        awaitUntilSlaveOK();
+        // sleep a while to recover
+        Thread.sleep(20000);
+    }
+
    @Test
    public void testRemoteActing_delayMsg() throws Exception {
        awaitUntilSlaveOK();
@@ -217,4 +276,83 @@ public class ScheduleSlaveActingMasterIT extends ContainerIntegrationTestBase {
        Thread.sleep(30000);
    }

+    @Test
+    public void testRemoteActing_timerMsg() throws Exception {
+        awaitUntilSlaveOK();
+
+        String topic = ScheduleSlaveActingMasterIT.class.getSimpleName() + random.nextInt(65535);
+        createTopic(topic);
+        AtomicInteger receivedMsgCount = new AtomicInteger(0);
+        AtomicInteger inTimeMsgCount = new AtomicInteger(0);
+        AtomicInteger master3MsgCount = new AtomicInteger(0);
+
+        MessageQueue messageQueue = new MessageQueue(topic, master1With3Replicas.getBrokerConfig().getBrokerName(), 0);
+        int sendSuccess = 0;
+        for (int i = 0; i < MESSAGE_COUNT; i++) {
+            Message msg = new Message(topic, MESSAGE_BODY);
+            msg.setDelayTimeSec(30);
+            SendResult sendResult = producer.send(msg, messageQueue);
+            if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
+                sendSuccess++;
+            }
+        }
+        final int finalSendSuccess = sendSuccess;
+        await().atMost(Duration.ofMinutes(1)).until(() -> finalSendSuccess >= MESSAGE_COUNT);
+        long sendCompleteTimeStamp = System.currentTimeMillis();
+        System.out.printf("send success%n");
+
+        DefaultMQPushConsumer pushConsumer = createPushConsumer(CONSUME_GROUP);
+        pushConsumer.subscribe(topic, "*");
+        pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+            long period = System.currentTimeMillis() - sendCompleteTimeStamp;
+            // Remote Acting lead to born timestamp, msgId changed, it need to polish.
+            if (Math.abs(period - 30000) <= 3000) {
+                inTimeMsgCount.addAndGet(msgs.size());
+            }
+            if (msgs.get(0).getBrokerName().equals(master3With3Replicas.getBrokerConfig().getBrokerName())) {
+                master3MsgCount.addAndGet(msgs.size());
+            }
+            receivedMsgCount.addAndGet(msgs.size());
+            msgs.forEach(x -> System.out.printf("cost " + period + " " + x + "%n"));
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+        pushConsumer.start();
+
+        isolateBroker(master1With3Replicas);
+        brokerContainer1.removeBroker(new BrokerIdentity(
+            master1With3Replicas.getBrokerConfig().getBrokerClusterName(),
+            master1With3Replicas.getBrokerConfig().getBrokerName(),
+            master1With3Replicas.getBrokerConfig().getBrokerId()));
+        System.out.printf("Remove master1%n");
+
+        isolateBroker(master2With3Replicas);
+        brokerContainer2.removeBroker(new BrokerIdentity(
+            master2With3Replicas.getBrokerConfig().getBrokerClusterName(),
+            master2With3Replicas.getBrokerConfig().getBrokerName(),
+            master2With3Replicas.getBrokerConfig().getBrokerId()));
+        System.out.printf("Remove master2%n");
+
+        await().atMost(Duration.ofMinutes(1)).until(() -> receivedMsgCount.get() >= MESSAGE_COUNT && master3MsgCount.get() >= MESSAGE_COUNT && inTimeMsgCount.get() >= MESSAGE_COUNT * 0.95);
+
+        System.out.printf("consumer received %d msg, %d in time%n", receivedMsgCount.get(), inTimeMsgCount.get());
+
+        pushConsumer.shutdown();
+
+        //Add back master
+        master1With3Replicas = brokerContainer1.addBroker(master1With3Replicas.getBrokerConfig(), master1With3Replicas.getMessageStoreConfig());
+        master1With3Replicas.start();
+        cancelIsolatedBroker(master1With3Replicas);
+        System.out.printf("Add back master1%n");
+
+        //Add back master
+        master2With3Replicas = brokerContainer2.addBroker(master2With3Replicas.getBrokerConfig(), master2With3Replicas.getMessageStoreConfig());
+        master2With3Replicas.start();
+        cancelIsolatedBroker(master2With3Replicas);
+        System.out.printf("Add back master2%n");
+
+        awaitUntilSlaveOK();
+        // sleep a while to recover
+        Thread.sleep(20000);
+    }
+
 }
@@ -96,7 +96,7 @@ public class ScheduledMessageIT extends ContainerIntegrationTestBase {
                inTimeMsgCount.addAndGet(msgs.size());
            }
            receivedMsgCount.addAndGet(msgs.size());
-            msgs.forEach(x -> System.out.printf(receivedMsgCount.get()+" cost " + period + " " + x + "%n"));
+            msgs.forEach(x -> System.out.printf(receivedMsgCount.get() + " cost " + period + " " + x + "%n"));

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
@@ -150,4 +150,38 @@ public class ScheduledMessageIT extends ContainerIntegrationTestBase {
            .until(() -> ((DefaultMessageStore) master1With3Replicas.getMessageStore()).getHaService().getConnectionCount().get() == 2);
    }

+    @Test
+    public void consumeTimerMsgFromSlave() throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
+        String topic = TOPIC_PREFIX + random.nextInt(65535);
+        createTopic(topic);
+        DefaultMQPushConsumer pushConsumer = createPushConsumer(CONSUME_GROUP);
+        pushConsumer.subscribe(topic, "*");
+        AtomicInteger receivedMsgCount = new AtomicInteger(0);
+        pushConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+            receivedMsgCount.addAndGet(msgs.size());
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+
+        for (int i = 0; i < MESSAGE_COUNT; i++) {
+            Message msg = new Message(topic, String.valueOf(i).getBytes());
+            msg.setDelayTimeSec(3);
+            producer.send(msg);
+        }
+
+        isolateBroker(master1With3Replicas);
+
+        producer.getDefaultMQProducerImpl().getmQClientFactory().updateTopicRouteInfoFromNameServer(topic);
+        assertThat(producer.getDefaultMQProducerImpl().getmQClientFactory().findBrokerAddressInPublish(topic)).isNull();
+
+        pushConsumer.start();
+
+        await().atMost(Duration.ofSeconds(MESSAGE_COUNT * 2)).until(() -> receivedMsgCount.get() >= MESSAGE_COUNT);
+
+        pushConsumer.shutdown();
+        cancelIsolatedBroker(master1With3Replicas);
+
+        await().atMost(100, TimeUnit.SECONDS)
+            .until(() -> ((DefaultMessageStore) master1With3Replicas.getMessageStore()).getHaService().getConnectionCount().get() == 2);
+    }
+
 }