opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-01 15:47:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7480] Fix the offset in the timerCheckPoint will not be corrected when the commitlog and consumeQueue are truncated (#7488)

Browse Source
This commit is contained in:
rongtong
2024-12-04 15:46:56 +08:00
committed by GitHub
parent e104c02d50
commit f4c498433d
1 changed files with 14 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/timer/TimerMessageStore.java
+14 -1
View File
@@ -293,6 +293,19 @@ public class TimerMessageStore {
}
currQueueOffset = Math.min(currQueueOffset, timerCheckpoint.getMasterTimerQueueOffset());
ConsumeQueueInterface cq = this.messageStore.getConsumeQueue(TIMER_TOPIC, 0);
// Correction based consume queue
if (cq != null && currQueueOffset < cq.getMinOffsetInQueue()) {
LOGGER.warn("Timer currQueueOffset:{} is smaller than minOffsetInQueue:{}",
currQueueOffset, cq.getMinOffsetInQueue());
currQueueOffset = cq.getMinOffsetInQueue();
} else if (cq != null && currQueueOffset > cq.getMaxOffsetInQueue()) {
LOGGER.warn("Timer currQueueOffset:{} is larger than maxOffsetInQueue:{}",
currQueueOffset, cq.getMaxOffsetInQueue());
currQueueOffset = cq.getMaxOffsetInQueue();
}
//check timer wheel
currReadTimeMs = timerCheckpoint.getLastReadTimeMs();
long nextReadTimeMs = formatTimeMs(
@@ -614,7 +627,7 @@ public class TimerMessageStore {
return;
}
if (msg.getProperty(TIMER_ENQUEUE_MS) != null
&& NumberUtils.toLong(msg.getProperty(TIMER_ENQUEUE_MS)) == Long.MAX_VALUE) {
&& NumberUtils.toLong(msg.getProperty(TIMER_ENQUEUE_MS)) == Long.MAX_VALUE) {
return;
}
// pass msg into addAndGet, for further more judgement extension.