[Bug] Timer dequeue may get stuck after CommitLog truncation due to stale TimerLog references

[chenxu80]

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.

• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.

• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

ubuntu 24.04

RocketMQ version

branch: develop version: 5.5.0 Git commit id: a6fb9e2

JDK Version

JDK 8u202

Describe the Bug

After a slave truncates its CommitLog to a smaller offset, Timer-related data structures such as TimerLog and TimerWheel may still retain records that reference CommitLog offsets beyond the truncation point.

These stale references can cause the Timer dequeue pipeline to read an unrelated message and retry it indefinitely.

Assume TimerLog contains a record like:

TimerLog entry -> CommitLog offset P

Originally, offset P contains a valid TIMER_TOPIC message.

After CommitLog is truncated to offset T, where T < P, the TimerLog entry is not removed. When CommitLog data is transferred again, offset P may be overwritten by a completely different message.

The resulting state is:

TimerLog entry -> CommitLog offset P
CommitLog[P]   -> non-TIMER_TOPIC message

When Timer dequeue processes this TimerLog entry, it reads the new message at P and treats it as the original Timer wrapper message.

The new message may not contain required Timer properties such as:

PROPERTY_REAL_TOPIC
PROPERTY_REAL_QUEUE_ID

This can cause an exception such as:

java.lang.NumberFormatException: null
    at java.lang.Integer.parseInt(Integer.java:542)
    at java.lang.Integer.parseInt(Integer.java:615)
    at org.apache.rocketmq.store.timer.TimerMessageStore.convertMessage(TimerMessageStore.java)
    at org.apache.rocketmq.store.timer.TimerMessageStore.convert(TimerMessageStore.java)
    at org.apache.rocketmq.store.timer.TimerMessageStore$TimerDequeuePutMessageService.run(TimerMessageStore.java)

With timerSkipUnknownError=false, TimerDequeuePutMessageService keeps retrying the same TimerRequest.

The TimerLog record is not repeatedly read. Instead, the already-created TimerRequest remains in the internal retry loop and fails continuously.

Impact

A single stale TimerLog reference can prevent the current Timer slot from completing.

As a result:

• Timer dequeue progress cannot advance.
• Subsequent scheduled messages cannot be delivered.
• Retry messages using the Timer mechanism, including POP revive retry delivery, may remain blocked.
• The broker continuously reports the same exception.

Steps to Reproduce

This issue can be reproduced with a simulated MessageStore without constructing the complete HA failure sequence.

1. Create a TimerMessageStore with a temporary TimerLog/TimerWheel directory and a mocked MessageStore.

2. Keep the default configuration:

   timerSkipUnknownError=false

3. Create a valid Timer message and append a TimerLog record that references a simulated CommitLog offset:

   offsetPy = P
   sizePy   = S
   

   Ensure the corresponding TimerWheel slot references this TimerLog record.

4. Configure the mocked MessageStore so that reading (P, S) now returns a valid ordinary message instead of the original TIMER_TOPIC message.

   The replacement message should:

   • Have a non-TIMER_TOPIC topic.
   • Not contain PROPERTY_REAL_TOPIC.
   • Not contain PROPERTY_REAL_QUEUE_ID.

   This simulates the state after CommitLog truncation and subsequent HA data overwrite.

5. Enable Timer dequeue and advance the Timer clock until the slot containing the simulated stale record is processed.

6. Run TimerDequeueGetMessageService and TimerDequeuePutMessageService in separate threads with a test timeout.

7. Observe that the replacement message is passed to convertMessage() and triggers:

   java.lang.NumberFormatException: null
       at java.lang.Integer.parseInt(Integer.java)
       at org.apache.rocketmq.store.timer.TimerMessageStore.convertMessage(TimerMessageStore.java)
       at org.apache.rocketmq.store.timer.TimerMessageStore.convert(TimerMessageStore.java)
       at org.apache.rocketmq.store.timer.TimerMessageStore$TimerDequeuePutMessageService.run(TimerMessageStore.java)
   

8. Verify that:

   • The same TimerRequest is retried repeatedly.
   • The request latch is not completed.
   • Timer read progress does not advance.
   • A valid Timer message scheduled after this record is not delivered.

9. Stop the Timer service explicitly at the end of the test to avoid leaving the retry thread running.

What Did You Expect to See?

The message referenced by TimerLog should be validated before conversion. If it is no longer a valid TIMER_TOPIC message, the stale request should be released and Timer dequeue should continue.

What Did You See Instead?

The ordinary message is treated as a Timer wrapper message. Conversion fails repeatedly, and the current Timer slot cannot complete.

Additional Context

No response

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Confirmed

The reported issue has been verified against the current codebase (develop branch).

Root Cause: DefaultMessageStore.truncateDirtyFiles() truncates CommitLog and consume queues but never cleans up TimerLog or TimerWheel. After truncation, stale TimerLog entries still reference offsets that may be overwritten by unrelated messages during re-replication.

Verified code path:

1. truncateDirtyFiles() (DefaultMessageStore.java:785-818) only truncates consume queues and CommitLog — no Timer cleanup
2. StoreUtil.getMessage() (StoreUtil.java:82-98) blindly decodes whatever bytes are at the stale offset
3. convertMessage() (TimerMessageStore.java:1266) calls Integer.parseInt(null) when PROPERTY_REAL_QUEUE_ID is missing → NumberFormatException
4. TimerDequeuePutMessageService.run() (TimerMessageStore.java:1622-1681) catches the exception, and with timerSkipUnknownError=false (default), retries the same TimerRequest indefinitely via holdMomentForUnknownError() (50ms sleep + loop)

Impact: Timer dequeue progress stalls completely. All subsequent scheduled messages and POP revive retry delivery remain blocked.

Severity: High — single stale reference can permanently block the Timer slot.

An automated fix proposal will be generated. Reply /approve to proceed with PR generation.

---

Automated evaluation by github-manager-bot

[RockteMQ-AI]

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.

Full Specification

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 TimerRequests 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 TimerRequests 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:

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.

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