Skip to content

[ISSUE #11156]fix(timer): persist TimelineRollService checkpoint to avoid repeated and loss #11157

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
3424672656 wants to merge 5 commits into
base: develop
Choose a base branch
from
Open

[ISSUE #11156]fix(timer): persist TimelineRollService checkpoint to avoid repeated and loss #11157

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5cdcf91
fix(timer): persist TimelineRollService checkpoint to avoid repeated …
Sep 12, 2026
76c5c71
Persist the roll checkpoint only after the messages are reput
Sep 13, 2026
e8c3aee
Drop timerRocksDBRollIntervalHours and back off on roll errors
Sep 13, 2026
33943ac
fix
Sep 13, 2026
b3b7490
Retrigger CI
Sep 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 0 additions & 9 deletions store/src/main/java/org/apache/rocketmq/store/config/MessageStoreConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
}
Expand Down
2 changes: 2 additions & 0 deletions store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
52 changes: 27 additions & 25 deletions store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

@RockteMQ-AI RockteMQ-AI Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Info] The hardcoded constants ROLL_TRIGGER_EARLY_MS = 1000L and ROLL_POLL_WHEN_NOT_DUE_MS = 1000L work well for the current use case. If these need tuning in production, consider exposing them via MessageStoreConfig with sensible defaults.

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);
Expand Down Expand Up @@ -373,37 +375,37 @@ 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);
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);
}

@RockteMQ-AI RockteMQ-AI Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Info] The checkpoint initialization logic is solid — starting from min(forwardCheckpoint, now) + maxDelay - rollRange ensures we don't miss messages that were already queued for expiration.

One consideration: if forwardCheckpoint is significantly behind now (e.g., after a long outage), the initial checkpoint could be far in the past. The service will catch up by scanning multiple windows in sequence, which is correct, but might want to log a warning when checkpoint < now - rollRange to make recovery visible.

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;

@RockteMQ-AI RockteMQ-AI Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Warning] The error handling loop waits only 200ms on failure before retrying. If scanRecordsToQueue consistently fails (e.g., RocksDB corruption, disk full), this could create a tight error loop.

Consider adding exponential backoff or a max retry count before logging at ERROR level and pausing longer:

int consecutiveErrors = 0;
// ...
if (!scanRecordsToQueue(...)) {
    consecutiveErrors++;
    long backoff = Math.min(200L * (1L << Math.min(consecutiveErrors, 5)), 10000L);
    this.waitForRunning(backoff);
    if (consecutiveErrors > 10) {
        logError.error("TimelineRollService: {} consecutive failures, backing off {}ms", consecutiveErrors, backoff);
    }
    continue;
}
consecutiveErrors = 0;

}
Comment thread
3424672656 marked this conversation as resolved.
} 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");
Expand Down
14 changes: 7 additions & 7 deletions store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
}
Expand Down Expand Up @@ -506,7 +506,7 @@ private class TimerMessageReputService extends ServiceThread {
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(),
Expand All @@ -518,10 +518,10 @@ private class TimerMessageReputService extends ServiceThread {
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
Expand All @@ -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) {

@RockteMQ-AI RockteMQ-AI Sep 14, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Critical] The null check null != this.checkPointKey is correct, but the Yoda condition style (null != x instead of x != null) is inconsistent with the rest of the codebase. RocketMQ style typically uses this.checkPointKey != null.

Minor, but worth aligning with project conventions for readability.

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());
Expand Down
60 changes: 60 additions & 0 deletions store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +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;

Expand Down Expand Up @@ -140,4 +146,58 @@ 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 = 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));
}
Comment thread
3424672656 marked this conversation as resolved.

@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);
}

}
Loading