RockteMQ-AI commented on PR #11157:
URL: https://github.com/apache/rocketmq/pull/11157#issuecomment-5647210682
### 🔍 Code Review — PR #11157
bash: line 180: develop: command not found
bash: line 180: TimelineRollService: command not found
bash: line 180: [now: command not found
bash: line 180: TIMER_TOPIC: command not found
bash: command substitution: line 181: syntax error: unexpected end of file
bash: line 180: timeline_roll_checkpoint: command not found
bash: command substitution: line 180: syntax error near unexpected token `)'
bash: command substitution: line 180: `[checkpoint, checkpoint + interval)'
bash: line 180: checkpoint: command not found
bash: line 180: MessageRocksDBStorageTest#testWriteAndGetRollCheckpoint:
command not found
bash: line 180: MessageRocksDBStorageTest#testScanAdjacentWindowsNoOverlap:
command not found
bash: line 180: timerMaxDelaySec: command not found
**Overall**
The change correctly moves the roll progress from memory to a RocksDB
checkpoint, which is the right direction for avoiding duplicate scans across
restarts. However, the checkpoint is advanced **before** the rolled records are
actually written back to the commitlog by the async `rollMessageReputService`,
so a crash can skip a window that was queued but not yet consumed. I think that
needs to be fixed before merging.
---
### 1. Correctness
**Persisting the checkpoint too early — likely data-loss/late-roll risk**
`store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:390-392`
```java
if (!scanRecordsToQueue(checkpoint, rangeMs,
timerMessageRocksDBStore.getRollMessageQueue())) {
...
}
checkpoint += rangeMs;
messageRocksDBStorage.writeCheckPointForTimer(...);
```
`scanRecordsToQueue` only **offers** the records to `rollMessageQueue`. The
actual re-put into the commitlog happens asynchronously in
`TimerMessageReputService`. If the broker crashes after the checkpoint is
persisted but before the queue consumer finishes, the next start will skip that
window. The records are still in RocksDB and will eventually be delivered by
`TimelineForwardService`, but the early-roll guarantee is broken and the PR’s
own test claim (“continue from the checkpoint without skipping … the previous
window”) is violated.
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.
**Initial checkpoint uses `rollRangeHour` while the scan window is
`rollIntervalHour`**
`store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:383-384`
```java
if (checkpoint <= 0L) {
checkpoint = System.currentTimeMillis() + maxDelayMs -
TimeUnit.HOURS.toMillis(rollRangeHour);
}
```
The window width is `rangeMs = rollIntervalHour`, but the initial offset is
`rollRangeHour`. If an operator configures `rollRangeHour < rollIntervalHour`,
the first upper bound becomes `now + maxDelay + (rollIntervalHour -
rollRangeHour)`, i.e. messages that are still farther than `maxDelay` away can
be rolled too early. At best this wastes work; at worst it can roll a message
that the timer wheel cannot yet hold.
Please validate that `rollRangeHour >= rollIntervalHour`, or initialize with
`rangeMs` instead.
**No backoff on unexpected exception**
`store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:395-397`
The outer `catch (Exception e)` logs and loops immediately. If RocksDB
throws a persistent error the thread will spin. Add a short `waitForRunning` in
the error path.
---
### 2. Performance
**Frequent checkpoint sync writes when catching up**
`store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:391`
`writeCheckPointForTimer` writes through the WAL. When the service is behind
it will scan consecutive 1-hour windows and persist after every single one. For
a large backlog this is a lot of small sync writes. Consider persisting only
every N windows or batching the advance.
**Hot loop on `storeConfig.isTimerStopDequeue()`**
`store/src/main/java/org/apache/rocketmq/store/timer/rocksdb/Timeline.java:388`
`scanRecordsToQueue` returns `false` immediately when dequeue is stopped,
then the new code waits only 200 ms. That is fine, but the log level is `error`
for an intentional pause — should be `warn` or `info`.
---
### 3. Tests
**No unit tests for `TimelineRollService` scheduling logic**
The PR only adds:
- `testWriteAndGetRollCheckpoint` — validates the checkpoint key read/write.
- `testScanAdjacentWindowsNoOverlap` — validates `[lower, upper)` scan
semantics.
Missing coverage:
- service waits when the next window is not yet due,
- service scans consecutive windows when behind,
- checkpoint is persisted and resumed on restart,
- no duplicate scan of the same window.
Please add a test for `TimelineRollService.run()` using a mocked
`MessageRocksDBStorage` and a fake clock, or extract the scheduling into a
package-private method.
**New test is flaky because it reuses global delay-time windows**
`store/src/test/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorageTest.java:155-178`
`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 the
exact-size assertions depending on execution order/time. Use isolated,
far-future delay times or clean the column family between tests.
**Pasted diff artifact**
The diff block shows lines like:
```java
@auth/src/test/java/org/apache/rocketmq/auth/authentication/AuthenticationEvaluatorTest.java
public void testWriteAndGetRollCheckpoint() {
```
Assuming this is a paste/rendering artifact, please confirm the real file
uses `@Test`.
---
### 4. Compatibility
**Behavioral change of `timerRocksDBRollIntervalHours` /
`timerRocksDBRollRangeHours`**
Previously `rollIntervalHours` controlled sleep and `rollRangeHours`
controlled the scan start offset, while the scan width was always
`timerMaxDelaySec`. Now `rollIntervalHours` is also the scan window width, and
`rollRangeHours` only affects the first checkpoint. Existing deployments that
tuned these configs will see different roll frequency and load patterns. Please
update the config documentation and consider a release note.
**Downgrade path**
`store/src/main/java/org/apache/rocketmq/store/rocksdb/MessageRocksDBStorage.java:75`
The new `TIMELINE_ROLL_CHECK_POINT` key is harmless to older code (it
ignores unknown keys), but an older broker restarting after a new broker has
advanced the checkpoint will not read it and will rescan already-rolled
windows, potentially producing duplicate timer messages. That may be
acceptable, but it should be documented.
---
**Recommendation:** Request changes. The checkpoint-before-consumption issue
and the test flakiness should be addressed; the config compatibility note
should be documented.
---
*Automated review by github-manager bot. Please verify suggestions before
applying.*
--
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]
