From 5cdcf91cbf95efbd3f880ee48a0d70a4a0a41abe Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sat, 12 Sep 2026 22:51:28 +0800 Subject: [PATCH 1/7] fix(timer): persist TimelineRollService checkpoint to avoid repeated rolls Scan hour-sized windows from a RocksDB checkpoint instead of sleeping a fixed interval, so delayed scans do not skip or re-roll the same timer messages. Co-authored-by: Cursor --- .../store/rocksdb/MessageRocksDBStorage.java | 2 + .../store/timer/rocksdb/Timeline.java | 49 ++++++++++--------- .../rocksdb/MessageRocksDBStorageTest.java | 45 +++++++++++++++++ 3 files changed, 72 insertions(+), 24 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java b/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java index d55596a293c..029cc1883dc 100644 --- a/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java +++ b/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java @@ -73,9 +73,11 @@ public class MessageRocksDBStorage extends AbstractRocksDBStorage { private static final Set COMMON_CHECK_POINT_KEY_SET_FOR_TIMER = new HashSet<>(); public static final byte[] SYS_TOPIC_SCAN_OFFSET_CHECK_POINT = "sys_topic_scan_offset_checkpoint".getBytes(StandardCharsets.UTF_8); public static final byte[] TIMELINE_CHECK_POINT = "timeline_checkpoint".getBytes(StandardCharsets.UTF_8); + public static final byte[] TIMELINE_ROLL_CHECK_POINT = "timeline_roll_checkpoint".getBytes(StandardCharsets.UTF_8); static { COMMON_CHECK_POINT_KEY_SET_FOR_TIMER.add(SYS_TOPIC_SCAN_OFFSET_CHECK_POINT); COMMON_CHECK_POINT_KEY_SET_FOR_TIMER.add(TIMELINE_CHECK_POINT); + COMMON_CHECK_POINT_KEY_SET_FOR_TIMER.add(TIMELINE_ROLL_CHECK_POINT); } private static final byte[] DELETE_VAL_FLAG = new byte[] {(byte)0xFF}; private static final int LAST_OFFSET_PY_LENGTH = LAST_OFFSET_PY.length; diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java index 740d5602b21..a901ed2aaac 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java @@ -49,6 +49,8 @@ public class Timeline { private static final String DELETE_KEY_SPLIT = "+"; private static final int ORIGIN_CAPACITY = 100000; private static final int BATCH_SIZE = 1000, MAX_BATCH_SIZE_FROM_ROCKSDB = 8000; + private static final long ROLL_TRIGGER_EARLY_MS = 1000L; + private static final long ROLL_POLL_WHEN_NOT_DUE_MS = 1000L; private static final int INITIAL = 0, RUNNING = 1, SHUTDOWN = 2; private volatile int state = INITIAL; private final AtomicLong commitOffset = new AtomicLong(0); @@ -373,35 +375,34 @@ public String getServiceName() { @Override public void run() { - log.info(this.getServiceName() + " service start"); + long checkpoint = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT); + log.info(this.getServiceName() + " service start, checkpoint: {}", checkpoint); while (!this.isStopped()) { - int rollIntervalHour = 1; - int rollRangeHour = 2; try { - if (storeConfig.getTimerRocksDBRollIntervalHours() > 0) { - rollIntervalHour = storeConfig.getTimerRocksDBRollIntervalHours(); - } - if (storeConfig.getTimerRocksDBRollRangeHours() > 0) { - rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours(); + long maxDelayMs = TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()); + int rollIntervalHour = storeConfig.getTimerRocksDBRollIntervalHours() > 0 ? storeConfig.getTimerRocksDBRollIntervalHours() : 1; + int rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2; + long rangeMs = TimeUnit.HOURS.toMillis(rollIntervalHour); + if (checkpoint <= 0L) { + checkpoint = System.currentTimeMillis() + maxDelayMs - TimeUnit.HOURS.toMillis(rollRangeHour); } - this.waitForRunning(TimeUnit.HOURS.toMillis(rollIntervalHour)); - if (stopped) { - log.info(this.getServiceName() + " service end"); - return; + long nextDueMs = checkpoint + rangeMs - maxDelayMs; + long triggerAt = nextDueMs - ROLL_TRIGGER_EARLY_MS; + long now = System.currentTimeMillis(); + if (now < triggerAt) { + this.waitForRunning(Math.min(triggerAt - now, ROLL_POLL_WHEN_NOT_DUE_MS)); + continue; } - } catch (Exception e) { - logError.error("Timeline TimelineRollService wait error: {}", e.getMessage()); - } - long rollCheckpoint = System.currentTimeMillis(); - try { - log.info("Timeline TimelineRollService start roll rollCheckpoint: {}", rollCheckpoint); - while (!scanRecordsToQueue(rollCheckpoint + TimeUnit.HOURS.toMillis(rollRangeHour), - TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()), - timerMessageRocksDBStore.getRollMessageQueue())) { - logError.error("Timeline TimelineRollService scanRecordsToQueue error."); - Thread.sleep(200); + + log.info("Timeline TimelineRollService start roll checkpoint: {}, rangeMs: {}, nextDueMs: {}, delayMs: {}", checkpoint, rangeMs, nextDueMs, now - nextDueMs); + if (!scanRecordsToQueue(checkpoint, rangeMs, timerMessageRocksDBStore.getRollMessageQueue())) { + logError.error("Timeline TimelineRollService scanRecordsToQueue error, checkpoint: {}", checkpoint); + this.waitForRunning(200L); + continue; } - log.info("Timeline TimelineRollService roll records success, lastRollTime: {}, rollCheckpoint: {}, cost: {}", rollCheckpoint, rollCheckpoint, System.currentTimeMillis() - rollCheckpoint); + checkpoint += rangeMs; + messageRocksDBStorage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT, checkpoint); + log.info("Timeline TimelineRollService roll records success, checkpoint: {}, cost: {}", checkpoint, System.currentTimeMillis() - now); } catch (Exception e) { logError.error("Timeline TimelineRollService failed error: {}", e.getMessage()); } diff --git a/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java b/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java index d28ef19f54c..893f0348386 100644 --- a/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java @@ -29,6 +29,7 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT; import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; public class MessageRocksDBStorageTest { @@ -140,4 +141,48 @@ public void testDeleteThenUpdate() { Assert.assertEquals(0, recordCount); } + @Test + public void testWriteAndGetRollCheckpoint() { + Assert.assertEquals(0L, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + + long checkpoint = System.currentTimeMillis() + 3600000L; + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, checkpoint); + Assert.assertEquals(checkpoint, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + + long nextCheckpoint = checkpoint + 3600000L; + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, nextCheckpoint); + Assert.assertEquals(nextCheckpoint, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + } + + @Test + public void testScanAdjacentWindowsNoOverlap() { + long window = 3600000L; + long begin = (System.currentTimeMillis() / window) * window; + + writeTimerRecord(begin + 1, "roll-window-first", 11L, 111); + writeTimerRecord(begin + window, "roll-window-boundary", 22L, 222); + writeTimerRecord(begin + window + 1, "roll-window-second", 33L, 333); + + List firstWindow = storage.scanRecordsForTimer( + TIMER_COLUMN_FAMILY, begin, begin + window, 10, null); + Assert.assertNotNull(firstWindow); + Assert.assertEquals(1, firstWindow.size()); + Assert.assertEquals("roll-window-first", firstWindow.get(0).getUniqKey()); + + List secondWindow = storage.scanRecordsForTimer( + TIMER_COLUMN_FAMILY, begin + window, begin + 2 * window, 10, null); + Assert.assertNotNull(secondWindow); + Assert.assertEquals(2, secondWindow.size()); + Assert.assertEquals("roll-window-boundary", secondWindow.get(0).getUniqKey()); + Assert.assertEquals("roll-window-second", secondWindow.get(1).getUniqKey()); + } + + private void writeTimerRecord(long delayTime, String uniqKey, long offsetPy, int sizePy) { + TimerRocksDBRecord record = new TimerRocksDBRecord(delayTime, uniqKey, offsetPy, sizePy, 0L, null); + record.setActionFlag(TimerRocksDBRecord.TIMER_ROCKSDB_PUT); + List list = new ArrayList<>(); + list.add(record); + storage.writeRecordsForTimer(TIMER_COLUMN_FAMILY, list); + } + } From 76c5c710afca5fe8b4974d2a7536e0a65ca132e0 Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sun, 13 Sep 2026 12:08:22 +0800 Subject: [PATCH 2/7] Persist the roll checkpoint only after the messages are reput TimelineRollService used to write the checkpoint right after a scan, so a crash between the scan and the reput dropped the whole window. The reput service now owns the write and takes the key to write, which lets the roll queue persist its own checkpoint the same way the expired queue already does. Roll keeps its scan frontier in memory and advances it per window, as TimelineForwardService does. Without a persisted roll checkpoint the frontier is derived from the delivery checkpoint plus one maximum delay, so a broker that ran without this key re-scans its windows instead of skipping them. Co-authored-by: Cursor --- .../store/timer/rocksdb/Timeline.java | 12 ++++--- .../rocksdb/TimerMessageRocksDBStore.java | 14 ++++----- .../rocksdb/MessageRocksDBStorageTest.java | 31 ++++++++++++++----- 3 files changed, 37 insertions(+), 20 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java index a901ed2aaac..6f4e69a266e 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java @@ -376,16 +376,19 @@ public String getServiceName() { @Override public void run() { long checkpoint = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT); + if (checkpoint <= 0L) { + long now = System.currentTimeMillis(); + long forwardCheckpoint = messageRocksDBStorage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_CHECK_POINT); + int rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2; + checkpoint = (forwardCheckpoint > 0L ? Math.min(forwardCheckpoint, now) : now) + + TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()) - TimeUnit.HOURS.toMillis(rollRangeHour); + } log.info(this.getServiceName() + " service start, checkpoint: {}", checkpoint); while (!this.isStopped()) { try { long maxDelayMs = TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()); int rollIntervalHour = storeConfig.getTimerRocksDBRollIntervalHours() > 0 ? storeConfig.getTimerRocksDBRollIntervalHours() : 1; - int rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2; long rangeMs = TimeUnit.HOURS.toMillis(rollIntervalHour); - if (checkpoint <= 0L) { - checkpoint = System.currentTimeMillis() + maxDelayMs - TimeUnit.HOURS.toMillis(rollRangeHour); - } long nextDueMs = checkpoint + rangeMs - maxDelayMs; long triggerAt = nextDueMs - ROLL_TRIGGER_EARLY_MS; long now = System.currentTimeMillis(); @@ -401,7 +404,6 @@ public void run() { continue; } checkpoint += rangeMs; - messageRocksDBStorage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT, checkpoint); log.info("Timeline TimelineRollService roll records success, checkpoint: {}, cost: {}", checkpoint, System.currentTimeMillis() - now); } catch (Exception e) { logError.error("Timeline TimelineRollService failed error: {}", e.getMessage()); diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java index c48e177c9d2..85a0d02ec91 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java @@ -231,8 +231,8 @@ private void initService() { this.expiredMessageQueue = new LinkedBlockingDeque<>(TIME_UP_CAPACITY); this.rollMessageQueue = new LinkedBlockingDeque<>(ROLL_CAPACITY); } - this.expiredMessageReputService = new TimerMessageReputService(expiredMessageQueue, storeConfig.getTimerRocksDBTimeExpiredMaxTps(), true); - this.rollMessageReputService = new TimerMessageReputService(rollMessageQueue, storeConfig.getTimerRocksDBRollMaxTps(), false); + this.expiredMessageReputService = new TimerMessageReputService(expiredMessageQueue, storeConfig.getTimerRocksDBTimeExpiredMaxTps(), MessageRocksDBStorage.TIMELINE_CHECK_POINT); + this.rollMessageReputService = new TimerMessageReputService(rollMessageQueue, storeConfig.getTimerRocksDBRollMaxTps(), MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT); this.timeline = new Timeline(messageStore, messageRocksDBStorage, this, timerMetrics); this.timerSysTopicScanService = new TimerSysTopicScanService(); } @@ -506,7 +506,7 @@ private class TimerMessageReputService extends ServiceThread { private final Logger log = TimerMessageRocksDBStore.log; private final BlockingQueue> queue; private final RateLimiter rateLimiter; - private final boolean writeCheckPoint; + private final byte[] checkPointKey; private final ExecutorService executor = ThreadUtils.newThreadPoolExecutor( storeConfig.getTimerReputServiceCorePoolSize(), @@ -518,10 +518,10 @@ private class TimerMessageReputService extends ServiceThread { new ThreadPoolExecutor.CallerRunsPolicy() ); - public TimerMessageReputService(BlockingQueue> queue, double maxTps, boolean writeCheckPoint) { + public TimerMessageReputService(BlockingQueue> queue, double maxTps, byte[] checkPointKey) { this.queue = queue; this.rateLimiter = RateLimiter.create(maxTps); - this.writeCheckPoint = writeCheckPoint; + this.checkPointKey = checkPointKey; } @Override @@ -545,9 +545,9 @@ public void run() { } countDownLatch.await(); log.info("TimerMessageReputService reput messages to commitlog, cost: {}, trs size: {}, checkPoint: {}", System.currentTimeMillis() - start, trs.size(), trs.get(trs.size() - 1).getCheckPoint()); - if (this.writeCheckPoint && !CollectionUtils.isEmpty(trs) && trs.get(trs.size() - 1).getCheckPoint() > 0L) { + if (null != this.checkPointKey && !CollectionUtils.isEmpty(trs) && trs.get(trs.size() - 1).getCheckPoint() > 0L) { log.info("TimerMessageReputService reput messages to commitlog, checkPoint: {}", trs.get(trs.size() - 1).getCheckPoint()); - messageRocksDBStorage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, MessageRocksDBStorage.TIMELINE_CHECK_POINT, trs.get(trs.size() - 1).getCheckPoint()); + messageRocksDBStorage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, this.checkPointKey, trs.get(trs.size() - 1).getCheckPoint()); } } catch (Exception e) { logError.error("TimerMessageReputService error: {}", e.getMessage()); diff --git a/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java b/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java index 893f0348386..73c82bf674f 100644 --- a/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java +++ b/store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java @@ -29,11 +29,16 @@ import java.util.ArrayList; import java.util.List; +import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMELINE_CHECK_POINT; import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMELINE_ROLL_CHECK_POINT; import static org.apache.rocketmq.store.rocksdb.MessageRocksDBStorage.TIMER_COLUMN_FAMILY; public class MessageRocksDBStorageTest { + /** Fixed delay time so window assertions never depend on the wall clock. */ + private static final long FIXED_DELAY_TIME_BASE = 2000000000000L; + private static final long WINDOW = 3600000L; + private MessageRocksDBStorage storage; private String storePath; @@ -145,32 +150,42 @@ public void testDeleteThenUpdate() { public void testWriteAndGetRollCheckpoint() { Assert.assertEquals(0L, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); - long checkpoint = System.currentTimeMillis() + 3600000L; + long checkpoint = FIXED_DELAY_TIME_BASE; storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, checkpoint); Assert.assertEquals(checkpoint, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); - long nextCheckpoint = checkpoint + 3600000L; + long nextCheckpoint = checkpoint + WINDOW; storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, nextCheckpoint); Assert.assertEquals(nextCheckpoint, storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); } + @Test + public void testRollCheckpointIsIndependentOfForwardCheckpoint() { + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_CHECK_POINT, FIXED_DELAY_TIME_BASE); + storage.writeCheckPointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT, FIXED_DELAY_TIME_BASE + WINDOW); + + Assert.assertEquals(FIXED_DELAY_TIME_BASE, + storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_CHECK_POINT)); + Assert.assertEquals(FIXED_DELAY_TIME_BASE + WINDOW, + storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT)); + } + @Test public void testScanAdjacentWindowsNoOverlap() { - long window = 3600000L; - long begin = (System.currentTimeMillis() / window) * window; + long begin = FIXED_DELAY_TIME_BASE; writeTimerRecord(begin + 1, "roll-window-first", 11L, 111); - writeTimerRecord(begin + window, "roll-window-boundary", 22L, 222); - writeTimerRecord(begin + window + 1, "roll-window-second", 33L, 333); + writeTimerRecord(begin + WINDOW, "roll-window-boundary", 22L, 222); + writeTimerRecord(begin + WINDOW + 1, "roll-window-second", 33L, 333); List firstWindow = storage.scanRecordsForTimer( - TIMER_COLUMN_FAMILY, begin, begin + window, 10, null); + TIMER_COLUMN_FAMILY, begin, begin + WINDOW, 10, null); Assert.assertNotNull(firstWindow); Assert.assertEquals(1, firstWindow.size()); Assert.assertEquals("roll-window-first", firstWindow.get(0).getUniqKey()); List secondWindow = storage.scanRecordsForTimer( - TIMER_COLUMN_FAMILY, begin + window, begin + 2 * window, 10, null); + TIMER_COLUMN_FAMILY, begin + WINDOW, begin + 2 * WINDOW, 10, null); Assert.assertNotNull(secondWindow); Assert.assertEquals(2, secondWindow.size()); Assert.assertEquals("roll-window-boundary", secondWindow.get(0).getUniqKey()); From e8c3aee86bd8f77c869eed6c11d4470cdf38fe5a Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sun, 13 Sep 2026 12:32:36 +0800 Subject: [PATCH 3/7] Drop timerRocksDBRollIntervalHours and back off on roll errors A checkpoint driven loop advances one window per window width, so the width and the cadence are the same quantity and only one config can define it. Keep timerRocksDBRollRangeHours as the scan window width and remove timerRocksDBRollIntervalHours, which no longer has a role of its own. The outer catch used to log and retry immediately, spinning the thread while rocksdb keeps throwing. Co-authored-by: Cursor --- .../apache/rocketmq/store/config/MessageStoreConfig.java | 9 --------- .../apache/rocketmq/store/timer/rocksdb/Timeline.java | 5 +++-- 2 files changed, 3 insertions(+), 11 deletions(-) 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 f0367023ddd..48122d64c7d 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 @@ -112,7 +112,6 @@ public class MessageStoreConfig { private long timerRocksDBPrecisionMs = 1000L; private double timerRocksDBRollMaxTps = 8000.0; private double timerRocksDBTimeExpiredMaxTps = 200000.0; - private int timerRocksDBRollIntervalHours = 1; private int timerRocksDBRollRangeHours = 2; private boolean timerRecallToTimeWheelEnable = true; private boolean timerRecallToTimelineEnable = true; @@ -2398,14 +2397,6 @@ public int getTimerReputServiceQueueCapacity() { return timerReputServiceQueueCapacity; } - public int getTimerRocksDBRollIntervalHours() { - return timerRocksDBRollIntervalHours; - } - - public void setTimerRocksDBRollIntervalHours(int timerRocksDBRollIntervalHours) { - this.timerRocksDBRollIntervalHours = timerRocksDBRollIntervalHours; - } - public int getTimerRocksDBRollRangeHours() { return timerRocksDBRollRangeHours; } diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java index 6f4e69a266e..98075336acb 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java @@ -387,8 +387,8 @@ public void run() { while (!this.isStopped()) { try { long maxDelayMs = TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()); - int rollIntervalHour = storeConfig.getTimerRocksDBRollIntervalHours() > 0 ? storeConfig.getTimerRocksDBRollIntervalHours() : 1; - long rangeMs = TimeUnit.HOURS.toMillis(rollIntervalHour); + int rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2; + long rangeMs = TimeUnit.HOURS.toMillis(rollRangeHour); long nextDueMs = checkpoint + rangeMs - maxDelayMs; long triggerAt = nextDueMs - ROLL_TRIGGER_EARLY_MS; long now = System.currentTimeMillis(); @@ -407,6 +407,7 @@ public void run() { log.info("Timeline TimelineRollService roll records success, checkpoint: {}, cost: {}", checkpoint, System.currentTimeMillis() - now); } catch (Exception e) { logError.error("Timeline TimelineRollService failed error: {}", e.getMessage()); + this.waitForRunning(200L); } } log.info(this.getServiceName() + " service end"); From 33943acbefd439b2a8b32e0bb204d7c823cf0700 Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sun, 13 Sep 2026 12:46:50 +0800 Subject: [PATCH 4/7] fix --- .../apache/rocketmq/store/timer/rocksdb/Timeline.java | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java index 98075336acb..18ab52342f3 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java @@ -387,17 +387,15 @@ public void run() { while (!this.isStopped()) { try { long maxDelayMs = TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()); - int rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2; - long rangeMs = TimeUnit.HOURS.toMillis(rollRangeHour); - long nextDueMs = checkpoint + rangeMs - maxDelayMs; - long triggerAt = nextDueMs - ROLL_TRIGGER_EARLY_MS; + long rangeMs = TimeUnit.HOURS.toMillis(storeConfig.getTimerRocksDBRollRangeHours() > 0 ? storeConfig.getTimerRocksDBRollRangeHours() : 2); + long triggerAt = checkpoint + rangeMs - maxDelayMs - ROLL_TRIGGER_EARLY_MS; long now = System.currentTimeMillis(); if (now < triggerAt) { - this.waitForRunning(Math.min(triggerAt - now, ROLL_POLL_WHEN_NOT_DUE_MS)); + this.waitForRunning(ROLL_POLL_WHEN_NOT_DUE_MS); continue; } - log.info("Timeline TimelineRollService start roll checkpoint: {}, rangeMs: {}, nextDueMs: {}, delayMs: {}", checkpoint, rangeMs, nextDueMs, now - nextDueMs); + log.info("Timeline TimelineRollService start roll checkpoint: {}, rangeMs: {}, triggerAt: {}, delayMs: {}", checkpoint, rangeMs, triggerAt, now - triggerAt); if (!scanRecordsToQueue(checkpoint, rangeMs, timerMessageRocksDBStore.getRollMessageQueue())) { logError.error("Timeline TimelineRollService scanRecordsToQueue error, checkpoint: {}", checkpoint); this.waitForRunning(200L); From b3b74908ce56a0c3f2eee8b30d632895002fdc1d Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sun, 13 Sep 2026 21:00:03 +0800 Subject: [PATCH 5/7] Retrigger CI Co-authored-by: Cursor From 2ece07ec3bc09d522439ef4fa8ff9a53f798b03f Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sun, 20 Sep 2026 17:13:26 +0800 Subject: [PATCH 6/7] Honor timerEnableRetryUntilSuccess in the RocksDB timer reput path Reuse the file-wheel flag so a recoverable put keeps retrying instead of advancing the roll checkpoint after a fixed number of failures. Co-authored-by: Cursor --- .../store/timer/rocksdb/TimerMessageRocksDBStore.java | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java index 85a0d02ec91..260619989cb 100644 --- a/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java @@ -560,7 +560,7 @@ private void putMsgWithRetry(MessageExtBrokerInner msg) throws InterruptedExcept if (null == msg) { return; } - for (int retryCount = 0; !isStopped() && retryCount <= MAX_PUT_MSG_TIMES; retryCount++) { + for (int retryCount = 0; !isStopped(); retryCount++) { int result = doPut(msg); switch (result) { case PUT_OK: @@ -569,13 +569,12 @@ private void putMsgWithRetry(MessageExtBrokerInner msg) throws InterruptedExcept logError.warn("Skipping message due to unrecoverable error. Msg: {}", msg); return; default: - if (retryCount == MAX_PUT_MSG_TIMES) { + if (!storeConfig.isTimerEnableRetryUntilSuccess() && retryCount >= MAX_PUT_MSG_TIMES) { logError.error("Message processing failed after {} retries. Msg: {}", retryCount, msg); return; - } else { - Thread.sleep(100L); - logError.warn("Retrying to process message. Retry count: {}, Msg: {}", retryCount, msg); } + Thread.sleep(100L); + logError.warn("Retrying to process message. Retry count: {}, Msg: {}", retryCount, msg); } } } From a2728d71c7e242f2b43cd1c61d023736b94e6927 Mon Sep 17 00:00:00 2001 From: hqbfzwang Date: Sun, 20 Sep 2026 17:22:24 +0800 Subject: [PATCH 7/7] Retrigger CI Co-authored-by: Cursor