RockteMQ-AI commented on code in PR #11157:
URL: https://github.com/apache/rocketmq/pull/11157#discussion_r4001827622
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:
##########
@@ -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);
+ }
+ 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;
Review Comment:
**[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:
```java
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;
```
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:
##########
@@ -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);
+ }
Review Comment:
**[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.
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/TimerMessageRocksDBStore.java:
##########
@@ -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) {
Review Comment:
**[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.
##########
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;
Review Comment:
**[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
