RockteMQ-AI commented on issue #10668:
URL: https://github.com/apache/rocketmq/issues/10668#issuecomment-5110639375
**Fix Spec Generated (v1)**
Code base: `00e45b8a6db2` (develop)
# Fix Specification — Issue #10668 (apache/rocketmq)
**Title:** Timer state (TimerLog / TimerWheel / timer cursors) is not
corrected after CommitLog truncation, leaving stale CommitLog references that
stall/mislead Timer enqueue & dequeue.
---
## 1. Root Cause
### 1.1 Truncation never reaches the timer subsystem
`DefaultMessageStore.truncateDirtyFiles(long offsetToTruncate)`
(`store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java:785-818`)
performs:
1. `reputMessageService.shutdown()` (line 794)
2. `truncateDirtyLogicFiles(offsetToTruncate)` →
`consumeQueueStore.truncateDirty(phyOffset)` (lines 799, 821-823)
3. `commitLog.truncateDirtyFiles(offsetToTruncate)` (line 802)
4. `recoverTopicQueueTable()` (line 804), then restarts a fresh reput
service (lines 806-817)
The `timerMessageStore` field (DefaultMessageStore.java:169, setter at
:1118) is **never referenced** in this flow. There is no hook, listener, or
callback interface for truncation events. Runtime callers of truncation are the
HA auto-switch path:
- `AutoSwitchHAService.truncateInvalidMsg()` →
`defaultMessageStore.truncateDirtyFiles(reputFromOffset)`
(`AutoSwitchHAService.java:545`)
- `AutoSwitchHAClient` → `messageStore.truncateFiles(truncateOffset)`
(`AutoSwitchHAClient.java:462` → `DefaultMessageStore.truncateFiles`, :826-838)
### 1.2 Why stale state breaks the timer
`TimerMessageStore`
(`store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java`)
keeps three pieces of state that embed CommitLog physical offsets:
- **TimerLog** entries (`TimerLog.java:33-41`, UNIT_SIZE=48) persist
`offsetPy` (CommitLog phy offset) and `sizePy` per delayed message.
Append-only; no correction API.
- **TimerWheel** slots (`Slot.java:19-50`) store `firstPos`/`lastPos`
pointing into TimerLog; those TimerLog entries in turn hold the (possibly
stale) CommitLog offsets.
- **Cursors**: `currQueueOffset` (TIMER topic consume-queue offset) and
`currReadTimeMs`/`currWriteTimeMs`.
<details>
<summary>Full Specification</summary>
# Fix Specification — Issue #10668 (apache/rocketmq)
**Title:** Timer state (TimerLog / TimerWheel / timer cursors) is not
corrected after CommitLog truncation, leaving stale CommitLog references that
stall/mislead Timer enqueue & dequeue.
---
## 1. Root Cause
### 1.1 Truncation never reaches the timer subsystem
`DefaultMessageStore.truncateDirtyFiles(long offsetToTruncate)`
(`store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java:785-818`)
performs:
1. `reputMessageService.shutdown()` (line 794)
2. `truncateDirtyLogicFiles(offsetToTruncate)` →
`consumeQueueStore.truncateDirty(phyOffset)` (lines 799, 821-823)
3. `commitLog.truncateDirtyFiles(offsetToTruncate)` (line 802)
4. `recoverTopicQueueTable()` (line 804), then restarts a fresh reput
service (lines 806-817)
The `timerMessageStore` field (DefaultMessageStore.java:169, setter at
:1118) is **never referenced** in this flow. There is no hook, listener, or
callback interface for truncation events. Runtime callers of truncation are the
HA auto-switch path:
- `AutoSwitchHAService.truncateInvalidMsg()` →
`defaultMessageStore.truncateDirtyFiles(reputFromOffset)`
(`AutoSwitchHAService.java:545`)
- `AutoSwitchHAClient` → `messageStore.truncateFiles(truncateOffset)`
(`AutoSwitchHAClient.java:462` → `DefaultMessageStore.truncateFiles`, :826-838)
### 1.2 Why stale state breaks the timer
`TimerMessageStore`
(`store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java`)
keeps three pieces of state that embed CommitLog physical offsets:
- **TimerLog** entries (`TimerLog.java:33-41`, UNIT_SIZE=48) persist
`offsetPy` (CommitLog phy offset) and `sizePy` per delayed message.
Append-only; no correction API.
- **TimerWheel** slots (`Slot.java:19-50`) store `firstPos`/`lastPos`
pointing into TimerLog; those TimerLog entries in turn hold the (possibly
stale) CommitLog offsets.
- **Cursors**: `currQueueOffset` (TIMER topic consume-queue offset) and
`currReadTimeMs`/`currWriteTimeMs`.
After `truncateDirtyFiles(offset)`:
- CommitLog data at `[offset, oldMax)` is gone; the TIMER topic consume
queue is shortened by `consumeQueueStore.truncateDirty`.
- TimerLog/TimerWheel are untouched — they continue to reference CommitLog
offsets ≥ `offset`.
- `currQueueOffset` may now exceed the TIMER consume queue max offset.
`enqueue()` (TimerMessageStore.java:751-836) only clamps against **min** offset
(lines 762-766); there is no max-offset clamp outside of startup `recover()`
(lines 332-336). Enqueue then repeatedly calls `cq.iterateFrom(offset)` for a
nonexistent offset and returns `false` forever — enqueue is effectively stuck
and newly written timer messages after the truncation point are never picked up
until restart.
- On dequeue, `dequeue()` (lines 1015-1124) walks stale TimerLog chains and
submits `TimerRequest`s with invalid `offsetPy`.
`TimerDequeueGetMessageService` (lines 1694-1775) calls
`getMessageByCommitOffset(tr.getOffsetPy(), tr.getSizePy())` (lines 1159-1169),
which retries the read **3 times per request** before returning null (each
attempt logs `"Fail to read msg from commitLog"`). With densely populated slots
this produces sustained retry churn/log flood; the request is then dropped via
`tr.idempotentRelease()` (lines 1747-1751). Additionally, `dequeue()` blocks in
`checkDequeueLatch` (lines 985-1012, called at 1098/1108) while the get-service
grinds through the retries, so the read cursor advances slowly ("dequeue stuck
retrying invalid messages").
- Startup recovery cannot repair this either: `recover()` (lines 299-363)
only runs from `load()`, and `timerWheel.checkPhyPos()`
(`TimerWheel.java:326-344`) validates **TimerLog** positions, not the CommitLog
offsets embedded in TimerLog entries. `TimerCheckpoint`
(`TimerCheckpoint.java:39-43`) does not record any CommitLog watermark, so a
truncation that happened while running is invisible to the next recovery.
**Summary:** the truncation event is not propagated to `TimerMessageStore`,
and the timer read paths have no defensive bounds check against the CommitLog's
current `[minOffset, maxOffset)` window.
## 2. Fix Strategy
Two complementary layers: an event-driven correction (primary) and defensive
bounds checks (safety net).
### 2.1 Primary — notify TimerMessageStore from `truncateDirtyFiles`
Add a correction method to `TimerMessageStore`, e.g. `public void
onCommitLogDispatchTruncate(long offsetToTruncate)` and invoke it from
`DefaultMessageStore.truncateDirtyFiles()` after `recoverTopicQueueTable()`
(i.e. after CommitLog/CQ are consistent, ~line 804), guarded by `if
(this.timerMessageStore != null)`.
The method must, under the enqueue/dequeue locks already used by the timer
services:
1. **Pause** enqueue/dequeue processing (reuse the existing running-state
flags, e.g. temporarily clear `isRunningEnqueue`/`isRunningDequeue` semantics
or hold their locks).
2. **Drain in-flight queues** (`enqueuePutQueue`, `dequeueGetQueue`,
`dequeuePutQueue`) of `TimerRequest`s whose `offsetPy >= offsetToTruncate`,
releasing their latches via `idempotentRelease()` so `dequeue()` is not left
waiting on `checkDequeueLatch`.
3. **Clamp `currQueueOffset`** to `min(currQueueOffset,
cq.getMaxOffsetInQueue())` for the TIMER topic consume queue (mirror of the
startup logic at TimerMessageStore.java:332-336); update `commitQueueOffset`
accordingly and call `prepareTimerCheckPoint()` (lines 1942-1954) so the
checkpoint is durable.
4. TimerLog entries with stale `offsetPy` remain physically on disk
(append-only format); they are neutralized by layer 2.2 below rather than
rewritten.
### 2.2 Defensive — fast-fail on out-of-range CommitLog offsets
In `getMessageByCommitOffset` (TimerMessageStore.java:1159-1169): before the
3-attempt retry loop, check
`offsetPy < messageStore.getCommitLog().getMinOffset() || offsetPy + sizePy
> messageStore.getMaxPhyOffset()` → log once at WARN and return `null`
immediately. This removes the 3× retry amplification for permanently-gone
offsets in both `enqueue()` (line 784), `reviseQueueOffset()` (line 373), and
the dequeue get-service (line 1718), while preserving retries for transient
read failures on valid offsets.
In `enqueue()` (lines 762-766): add the symmetric max-offset clamp:
```java
if (currQueueOffset > cq.getMaxOffsetInQueue()) {
LOGGER.warn(...);
currQueueOffset = cq.getMaxOffsetInQueue();
}
```
so enqueue self-heals even if the notification path is bypassed (e.g. plugin
stores).
Dequeue-side behavior for null messages (drop + `idempotentRelease`, lines
1747-1751) is kept — with the fast-fail check it now completes promptly instead
of spinning through retries.
## 3. Files to Modify
| File | Change |
|---|---|
| `store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java` |
In `truncateDirtyFiles()` (785-818), after `recoverTopicQueueTable()`, invoke
`timerMessageStore.onCommitLogDispatchTruncate(offsetToTruncate)` when
non-null. |
|
`store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java` |
New `onCommitLogDispatchTruncate(long)`; bounds fast-fail in
`getMessageByCommitOffset()` (1159-1169); max-offset clamp in `enqueue()`
(after 766). |
|
`store/src/test/java/org/apache/rocketmq/store/timer/TimerMessageStoreTest.java`
| New tests (see §4). |
No changes needed to `TimerLog.java`, `TimerWheel.java`, `Slot.java`, or
`TimerCheckpoint.java` (on-disk formats untouched).
## 4. Test Plan
Extend `TimerMessageStoreTest` (pattern of `testStateAndRecover()`, lines
468-540):
1. **testDequeueAfterCommitLogTruncate** — put N timer messages, wait until
enqueued into TimerWheel; call `messageStore.truncateDirtyFiles(offset)`
cutting the last K messages; assert: (a) dequeue does not block —
`currReadTimeMs` keeps advancing past the affected slots within a bounded time;
(b) the surviving N−K messages are delivered; (c) no request remains stuck in
`dequeueGetQueue`/`dequeuePutQueue`.
2. **testEnqueueCursorClampAfterTruncate** — drive `currQueueOffset` past
the truncated CQ max, call truncation, then put a new timer message; assert
enqueue resumes and the new message is delivered (verifies the max clamp).
3. **testGetMessageByCommitOffsetFastFail** — unit-level: offset beyond
`getMaxPhyOffset()` returns null without 3 store reads (verify via
counter/mock).
4. **testRecoverAfterTruncateAndRestart** — truncate, shutdown, restart
`TimerMessageStore`; assert `recover()` (299-363) completes and
`currQueueOffset ≤ cq.getMaxOffsetInQueue()`.
5. **Regression** — run existing `TimerMessageStoreTest`, `TimerLogTest`,
`TimerWheelTest`, `TimerCheckPointTest`, and HA `AutoSwitchHAService`-related
tests.
## 5. Backward Compatibility
- **On-disk formats unchanged**: TimerLog unit layout (48B), TimerWheel slot
layout, and TimerCheckpoint layout are untouched — rolling upgrade/downgrade
safe.
- **Public API**: one new public method on `TimerMessageStore`; no interface
(`MessageStore`) signature changes, so plugin stores
(`AbstractPluginMessageStore`) are unaffected.
- **Behavioral change**: messages whose CommitLog data was truncated are now
skipped promptly instead of retried 3×; they were already unrecoverable (data
deleted), so no delivery semantics regress. Truncation on HA switch already
implies the truncated tail is invalid on this replica; the new master retains
authoritative copies.
- `timerMessageStore == null` (timer disabled, `timerWheelEnable=false`) —
guard keeps old behavior.
## 6. Risk Assessment
| Risk | Level | Mitigation |
|---|---|---|
| Deadlock between truncation thread and timer service threads while
pausing/draining | Medium | Use existing lock ordering of enqueue/dequeue
services; drain queues with `poll()` not blocking `take()`; add timeout + WARN
log. Covered by test 1. |
| Dropping an in-flight `TimerRequest` without releasing its latch →
`dequeue()` blocks in `checkDequeueLatch` (985-1012) | Medium | Always
`idempotentRelease()` drained requests (idempotent by design). |
| Fast-fail bounds check races with concurrent CommitLog append
(maxPhyOffset moving forward) | Low | Only *skips retry* when offset is beyond
max at check time; a message freshly appended is ≤ max by the time its
TimerRequest is created, and the check re-reads live offsets. Worst case falls
through to the existing 3-retry loop. |
| `currQueueOffset` clamp interacting with slave sync
(`masterTimerQueueOffset`, TimerCheckpoint.java:39-43) | Low | Clamp uses same
rule already applied in `recover()` (332-336); checkpoint refreshed via
`prepareTimerCheckPoint()`. |
| Missed correction on code paths not going through
`DefaultMessageStore.truncateDirtyFiles` | Low | Defensive layer (§2.2) makes
enqueue/dequeue self-healing regardless of notification. |
**Overall: Medium-low.** Changes are localized to the store module timer
path; the defensive layer alone eliminates the stuck/retry symptom, and the
notification layer restores cursor consistency without touching persisted
formats.
</details>
**Next Steps:**
- Reply `/approve` to proceed with PR generation
- Reply `/revise <feedback>` to request changes
- Reply `/reject` to close this proposal
*This proposal expires in 72 hours.*
---
*Automated by github-manager-bot*
--
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]
