This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new fb1b30cbd2 [ISSUE #11156]fix(timer): persist TimelineRollService
checkpoint to avoid repeated and loss (#11157)
fb1b30cbd2 is described below
commit fb1b30cbd2edeed713804d10f8cf0c12f80552a5
Author: hqbfz <[email protected]>
AuthorDate: Tue Sep 22 11:43:11 2026 +0800
[ISSUE #11156]fix(timer): persist TimelineRollService checkpoint to avoid
repeated and loss (#11157)
* 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 <[email protected]>
* 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 <[email protected]>
* 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 <[email protected]>
* fix
* Retrigger CI
Co-authored-by: Cursor <[email protected]>
* 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 <[email protected]>
* Retrigger CI
Co-authored-by: Cursor <[email protected]>
---------
Co-authored-by: hqbfzwang <[email protected]>
Co-authored-by: Cursor <[email protected]>
---
.../rocketmq/store/config/MessageStoreConfig.java | 9 ----
.../store/rocksdb/MessageRocksDBStorage.java | 2 +
.../rocketmq/store/timer/rocksdb/Timeline.java | 52 ++++++++++---------
.../timer/rocksdb/TimerMessageRocksDBStore.java | 23 ++++-----
.../store/rocksdb/MessageRocksDBStorageTest.java | 60 ++++++++++++++++++++++
5 files changed, 100 insertions(+), 46 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 f0367023dd..48122d64c7 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 class MessageStoreConfig {
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/rocksdb/MessageRocksDBStorage.java
b/store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java
index d55596a293..029cc1883d 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<byte[]> 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 740d5602b2..18ab52342f 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,37 +375,37 @@ public class Timeline {
@Override
public void run() {
- log.info(this.getServiceName() + " service start");
+ 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()) {
- int rollIntervalHour = 1;
- int rollRangeHour = 2;
try {
- if (storeConfig.getTimerRocksDBRollIntervalHours() > 0) {
- rollIntervalHour =
storeConfig.getTimerRocksDBRollIntervalHours();
- }
- if (storeConfig.getTimerRocksDBRollRangeHours() > 0) {
- rollRangeHour =
storeConfig.getTimerRocksDBRollRangeHours();
- }
-
this.waitForRunning(TimeUnit.HOURS.toMillis(rollIntervalHour));
- if (stopped) {
- log.info(this.getServiceName() + " service end");
- return;
+ long maxDelayMs =
TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec());
+ 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(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: {}, triggerAt: {}, delayMs: {}", checkpoint, rangeMs,
triggerAt, now - triggerAt);
+ 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;
+ 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");
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 c48e177c9d..260619989c 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 @@ public class TimerMessageRocksDBStore {
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 @@ public class TimerMessageRocksDBStore {
private final Logger log = TimerMessageRocksDBStore.log;
private final BlockingQueue<List<TimerRocksDBRecord>> 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 @@ public class TimerMessageRocksDBStore {
new ThreadPoolExecutor.CallerRunsPolicy()
);
- public
TimerMessageReputService(BlockingQueue<List<TimerRocksDBRecord>> queue, double
maxTps, boolean writeCheckPoint) {
+ public
TimerMessageReputService(BlockingQueue<List<TimerRocksDBRecord>> 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 class TimerMessageRocksDBStore {
}
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());
@@ -560,7 +560,7 @@ public class TimerMessageRocksDBStore {
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 @@ public class TimerMessageRocksDBStore {
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);
}
}
}
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 d28ef19f54..73c82bf674 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,10 +29,16 @@ import java.io.File;
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;
@@ -140,4 +146,58 @@ public class MessageRocksDBStorageTest {
Assert.assertEquals(0, recordCount);
}
+ @Test
+ public void testWriteAndGetRollCheckpoint() {
+ Assert.assertEquals(0L,
storage.getCheckpointForTimer(TIMER_COLUMN_FAMILY, TIMELINE_ROLL_CHECK_POINT));
+
+ 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 + 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 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);
+
+ List<TimerRocksDBRecord> 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<TimerRocksDBRecord> 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<TimerRocksDBRecord> list = new ArrayList<>();
+ list.add(record);
+ storage.writeRecordsForTimer(TIMER_COLUMN_FAMILY, list);
+ }
+
}
