[ISSUE #10073] Fix StoreCheckpoint logicsMsgTimestamp to reflect flushed consume queue storetime (#10074)

* Fix StoreCheckpoint: set logicsMsgTimestamp only after CQ flush, use logicsMsgTempTimestamp for in-memory storetime

Change-Id: I6085bf6efaef84168ece31d080481717465f2b13

* rename to tmpLogicsMsgTimestamp

Change-Id: Ia65ca06751f765bdc2bf053c58e08789f4b2fb22

---------

Co-authored-by: guyinyou <guyinyou.gyy@alibaba-inc.com>

Author: guyinyou

store/src/main/java/org/apache/rocketmq/store/ConsumeQueue.java

@@ -709,7 +709,7 @@ public void putMessagePositionInfoWrapper(DispatchRequest request) {
                     this.messageStore.getMessageStoreConfig().isEnableDLegerCommitLog()) {
                     this.messageStore.getStoreCheckpoint().setPhysicMsgTimestamp(request.getStoreTimestamp());
                 }
-                this.messageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
+                this.messageStore.getStoreCheckpoint().setTmpLogicsMsgTimestamp(request.getStoreTimestamp());
                 if (MultiDispatchUtils.checkMultiDispatchQueue(this.messageStore.getMessageStoreConfig(), request)) {
                     multiDispatchLmqQueue(request, maxRetries);
                 }

store/src/main/java/org/apache/rocketmq/store/StoreCheckpoint.java

@@ -33,6 +33,7 @@ public class StoreCheckpoint {
     private final RandomAccessFile randomAccessFile;
     private final FileChannel fileChannel;
     private final MappedByteBuffer mappedByteBuffer;
+    private volatile long tmpLogicsMsgTimestamp = 0;
     private volatile long physicMsgTimestamp = 0;
     private volatile long logicsMsgTimestamp = 0;
     private volatile long indexMsgTimestamp = 0;
@@ -112,6 +113,14 @@ public void setLogicsMsgTimestamp(long logicsMsgTimestamp) {
         this.logicsMsgTimestamp = logicsMsgTimestamp;
     }
 
+    public long getTmpLogicsMsgTimestamp() {
+        return tmpLogicsMsgTimestamp;
+    }
+
+    public void setTmpLogicsMsgTimestamp(long tmpLogicsMsgTimestamp) {
+        this.tmpLogicsMsgTimestamp = tmpLogicsMsgTimestamp;
+    }
+
     public long getConfirmPhyOffset() {
         return confirmPhyOffset;
     }

store/src/main/java/org/apache/rocketmq/store/queue/BatchConsumeQueue.java

@@ -536,7 +536,7 @@ public void putMessagePositionInfoWrapper(DispatchRequest request) {
                 if (BrokerRole.SLAVE == this.messageStore.getMessageStoreConfig().getBrokerRole()) {
                     this.messageStore.getStoreCheckpoint().setPhysicMsgTimestamp(request.getStoreTimestamp());
                 }
-                this.messageStore.getStoreCheckpoint().setLogicsMsgTimestamp(request.getStoreTimestamp());
+                this.messageStore.getStoreCheckpoint().setTmpLogicsMsgTimestamp(request.getStoreTimestamp());
                 return;
             } else {
                 // XXX: warn and notify me

store/src/main/java/org/apache/rocketmq/store/queue/ConsumeQueueStore.java

@@ -648,23 +648,25 @@ private void doFlush(int retryTimes) {
             if (currentTimeMillis >= (this.lastFlushTimestamp + flushConsumeQueueThoroughInterval)) {
                 this.lastFlushTimestamp = currentTimeMillis;
                 flushConsumeQueueLeastPages = 0;
-                logicsMsgTimestamp = messageStore.getStoreCheckpoint().getLogicsMsgTimestamp();
+                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;
                 }
             }
 
             if (messageStoreConfig.isEnableCompaction()) {
                 messageStore.getCompactionStore().flush(flushConsumeQueueLeastPages);
             }
 
-            if (0 == flushConsumeQueueLeastPages) {
+            if (flushOK && 0 == flushConsumeQueueLeastPages) {
                 if (logicsMsgTimestamp > 0) {
                     messageStore.getStoreCheckpoint().setLogicsMsgTimestamp(logicsMsgTimestamp);
                 }