[ISSUE #11156]fix(timer): persist TimelineRollService checkpoint to avoid repeated and loss - #11157
[ISSUE #11156]fix(timer): persist TimelineRollService checkpoint to avoid repeated and loss#111573424672656 wants to merge 5 commits into
Conversation
…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 <cursoragent@cursor.com>
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #11157 +/- ##
=============================================
- Coverage 49.39% 49.29% -0.11%
+ Complexity 14237 14210 -27
=============================================
Files 1390 1390
Lines 103123 103121 -2
Branches 13484 13486 +2
=============================================
- Hits 50940 50832 -108
- Misses 46031 46104 +73
- Partials 6152 6185 +33 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
🔍 Code Review — PR #11157bash: line 180: develop: command not found 1. CorrectnessPersisting the checkpoint too early — likely data-loss/late-roll risk if (!scanRecordsToQueue(checkpoint, rangeMs, timerMessageRocksDBStore.getRollMessageQueue())) {
...
}
checkpoint += rangeMs;
messageRocksDBStorage.writeCheckPointForTimer(...);
Consider advancing the checkpoint only after the roll reput service has finished processing the window, similar to how Initial checkpoint uses if (checkpoint <= 0L) {
checkpoint = System.currentTimeMillis() + maxDelayMs - TimeUnit.HOURS.toMillis(rollRangeHour);
}The window width is Please validate that No backoff on unexpected exception The outer 2. PerformanceFrequent checkpoint sync writes when catching up
Hot loop on
3. TestsNo unit tests for
Missing coverage:
Please add a test for New test is flaky because it reuses global delay-time windows
Pasted diff artifact @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 4. CompatibilityBehavioral change of Downgrade path The new 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. |
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR persists the TimelineRollService checkpoint to RocksDB to avoid duplicate scans and message loss across restarts. The approach is sound in principle, but the checkpoint is advanced before the async roll reput service finishes processing, which can cause window skipping on crash. Several other concerns around initial checkpoint alignment, error handling, and test isolation need attention.
Findings
- [Critical] Timeline.java:390 — Checkpoint advanced before async roll completes; crash can skip windows
- [Warning] Timeline.java:383 — Initial checkpoint uses rollRangeHour but window is rollIntervalHour; mismatch risk
- [Warning] Timeline.java:395 — No backoff on persistent RocksDB errors; thread will spin
- [Info] Timeline.java:391 — Consider batching checkpoint writes to reduce WAL overhead
- [Warning] MessageRocksDBStorageTest.java:155 — Test window boundaries may collide with other tests
Automated review by github-manager-bot
| if (storeConfig.getTimerRocksDBRollRangeHours() > 0) { | ||
| rollRangeHour = storeConfig.getTimerRocksDBRollRangeHours(); | ||
| long maxDelayMs = TimeUnit.SECONDS.toMillis(storeConfig.getTimerMaxDelaySec()); | ||
| int rollIntervalHour = storeConfig.getTimerRocksDBRollIntervalHours() > 0 ? storeConfig.getTimerRocksDBRollIntervalHours() : 1; |
There was a problem hiding this 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.
52d7457 to
5cdcf91
Compare
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 <cursoragent@cursor.com>
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 <cursoragent@cursor.com>
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR persists the TimelineRollService checkpoint to RocksDB to avoid duplicate scans and message loss across restarts. The updated version addresses all previously raised concerns:
- Checkpoint persistence timing ✅ — Checkpoint is now persisted by
TimerRollReputService.doReput()after records are successfully written to the commitlog, ensuring crash-safety. - Initial checkpoint alignment ✅ — Initialization properly calculates the starting checkpoint from
maxDelay, avoiding out-of-range windows on first run. - Error handling backoff ✅ —
Thread.sleep(200)added in both the scan-error and general-exception paths, preventing tight spin loops on persistent errors. - Test isolation ✅ — Tests use a fixed time base (
BASE_TIME_MS) instead ofSystem.currentTimeMillis(), ensuring deterministic behavior.
The removal of timerRocksDBRollIntervalHours simplifies the configuration model by deriving the roll cadence from rollRangeHours and the delay window. Note: this is a minor backward-incompatible config change — users who had customized this value will need to remove it from their configs.
Overall the approach is sound and the implementation is now correct. LGTM.
Automated review by github-manager-bot
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Re-reviewed after new commits. The checkpoint-driven roll approach is well-implemented — persisted RocksDB checkpoint replaces the fragile fixed-sleep logic, preventing both duplicate and lost timer message rolls across restarts.
Key changes look correct:
TIMELINE_ROLL_CHECK_POINTproperly registered inCOMMON_CHECK_POINT_KEY_SET_FOR_TIMER- Poll-based trigger with 1s early wake is a reasonable trade-off between precision and CPU
- Removal of
timerRocksDBRollIntervalHoursis consistent since the interval is now derived from the checkpoint
LGTM — re-approved with the latest changes.
Automated review by github-manager-bot
Co-authored-by: Cursor <cursoragent@cursor.com>
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR fixes a critical correctness issue in TimelineRollService where the fixed-interval sleep + wall-clock-based scan window could cause timer messages to be rolled multiple times or skipped after restarts. The checkpoint-driven approach with persisted RocksDB state is the right solution.
Findings
- [Critical] TimerMessageRocksDBStore.java:548 — Yoda condition style inconsistent with codebase
- [Warning] Timeline.java:395 — Tight error loop risk on persistent failures
- [Info] Timeline.java:385 — Checkpoint initialization could log recovery scenarios
- [Info] Timeline.java:52 — Hardcoded timing constants work but could be configurable
Strengths
✅ Checkpoint persistence eliminates duplicate/skipped rolls across restarts
✅ Adjacent window scanning logic is correct and well-tested
✅ Refactoring TimerMessageReputService to use checkpoint keys is clean and extensible
✅ Test coverage for checkpoint independence and window boundaries is solid
Suggestions
- Add exponential backoff for error retries to avoid tight loops
- Log a warning when initial checkpoint is far behind current time (recovery scenario)
- Align null-check style with codebase conventions
Overall, this is a well-designed fix for a subtle concurrency/persistence bug. The changes are focused and the test coverage is good.
| 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); | ||
| } |
There was a problem hiding this 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.
| long now = System.currentTimeMillis(); | ||
| if (now < triggerAt) { | ||
| this.waitForRunning(ROLL_POLL_WHEN_NOT_DUE_MS); | ||
| continue; |
There was a problem hiding this 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:
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;| 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; |
There was a problem hiding this 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.
| 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) { |
There was a problem hiding this 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.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR correctly addresses the issue of timer roll service repeating or losing rolls after restart by replacing the fixed-interval approach with checkpoint-based scheduling.
The implementation is clean:
- Removes the unused
timerRocksDBRollIntervalHoursconfig - Adds
TIMELINE_ROLL_CHECK_POINTfor persisting roll state - Refactors
TimelineRollServiceto use checkpoint-based scheduling - Updates
TimerMessageReputServiceto support flexible checkpoint keys - Includes comprehensive tests
LGTM. Well-structured fix with proper test coverage.
Automated review by github-manager
3 participants
Which Issue(s) This PR Fixes
Brief Description
What is the problem?
TimelineRollServiceused a fixed sleep plus a scan window of[now + rollRange, now + rollRange + timerMaxDelaySec].Two issues follow from that:
TIMER_TOPICmany times.System.currentTimeMillis()after sleep. If a scan is delayed, the next round can skip or rescan the same delayTime range. The progress was also only in memory, so a restart could roll the same messages again.What is the solution?
Drive roll by a persisted RocksDB checkpoint (
timeline_roll_checkpoint) instead of a fixed sleep:[checkpoint, checkpoint + interval).checkpoint + interval - timerMaxDelaySec. Trigger 1s early; if it is not due yet, poll with at most 1s wait.Test plan
MessageRocksDBStorageTest#testWriteAndGetRollCheckpointMessageRocksDBStorageTest#testScanAdjacentWindowsNoOverlaptimerMaxDelaySeca bit larger) and confirm it is rolled at most once