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

[ISSUE #10079] FlushConsumeQueueService: always flush store checkpoint after CQ flush (#10080)

Browse Source 
Change-Id: I57c0922bb81c2d43359867e82a92fdf2deab7ad7

Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>
This commit is contained in:
guyinyou
2026-02-09 17:38:03 +08:00
committed by GitHub
parent 1754cece4b
commit c197e50f8d
1 changed files with 1 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStore.java
+1 -3
View File
@@ -651,14 +651,12 @@ public class ConsumeQueueStore extends AbstractConsumeQueueStore {
logicsMsgTimestamp = messageStore.getStoreCheckpoint().getTmpLogicsMsgTimestamp();
}
boolean flushOK = true;
for (ConcurrentMap<Integer, ConsumeQueueInterface> maps : consumeQueueTable.values()) {
for (ConsumeQueueInterface cq : maps.values()) {
boolean result = false;
for (int i = 0; i < retryTimes && !result; i++) {
result = flush(cq, flushConsumeQueueLeastPages);
}
flushOK &= result;
}
}
@@ -666,7 +664,7 @@ public class ConsumeQueueStore extends AbstractConsumeQueueStore {
messageStore.getCompactionStore().flush(flushConsumeQueueLeastPages);
}
if (flushOK && 0 == flushConsumeQueueLeastPages) {
if (0 == flushConsumeQueueLeastPages) {
if (logicsMsgTimestamp > 0) {
messageStore.getStoreCheckpoint().setLogicsMsgTimestamp(logicsMsgTimestamp);
}