RockteMQ-AI commented on code in PR #11157:
URL: https://github.com/apache/rocketmq/pull/11157#discussion_r3997010472
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:
##########
@@ -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;
Review Comment:
**[Critical]** Checkpoint is advanced *before* the rolled records are
actually written back to the commitlog. `scanRecordsToQueue` only offers
records to `rollMessageQueue`; the actual re-put happens asynchronously in
`TimerMessageReputService`. If the broker crashes after checkpoint persist but
before the queue consumer finishes, the next start skips that window.
Consider advancing the checkpoint only after the roll reput service has
finished processing the window, similar to how
`TimerMessageReputService.writeCheckPoint` works for the expired queue.
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:
##########
@@ -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;
Review Comment:
**[Warning]** Initial checkpoint uses `rollRangeHour` while the scan window
is `rollIntervalHour` (`rangeMs`). If `rollRangeHour < rollIntervalHour`, the
first upper bound extends beyond `maxDelay`, potentially rolling messages too
early.
Please validate that `rollRangeHour >= rollIntervalHour`, or initialize with
`rangeMs` instead.
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:
##########
@@ -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();
Review Comment:
**[Info]** `writeCheckPointForTimer` writes through the WAL. When catching
up from a backlog, this persists after every 1-hour window. Consider persisting
only every N windows or batching the advance to reduce sync write overhead.
##########
store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java:
##########
@@ -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));
+ }
Review Comment:
**[Warning]** `testScanAdjacentWindowsNoOverlap` uses `begin =
(System.currentTimeMillis() / window) * window`. Other tests in the same class
write records near `currentTime + 3600000L`, which can fall into these windows
and break exact-size assertions depending on execution order. Use isolated,
far-future delay times or clean the column family between tests.
##########
store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:
##########
@@ -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;
}
Review Comment:
**[Warning]** The outer `catch (Exception e)` logs and loops immediately. If
RocksDB throws a persistent error, the thread will spin. Add a short
`waitForRunning` backoff in the error path.
--
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]
