[ISSUE #9371] Delete ConsumeQueue index before CommitLog in tiered storage (#9372)

Author: lizhimins

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java

@@ -124,17 +124,17 @@ public List<FileSegment> getFileSegmentList() {
 
     public long getMinOffset() {
         List<FileSegment> list = this.fileSegmentTable;
-        return list.isEmpty() ? GET_FILE_SIZE_ERROR : list.get(0).getBaseOffset();
+        return list.isEmpty() ? 0L : list.get(0).getBaseOffset();
     }
 
     public long getCommitOffset() {
         List<FileSegment> list = this.fileSegmentTable;
-        return list.isEmpty() ? GET_FILE_SIZE_ERROR : list.get(list.size() - 1).getCommitOffset();
+        return list.isEmpty() ? 0L : list.get(list.size() - 1).getCommitOffset();
     }
 
     public long getAppendOffset() {
         List<FileSegment> list = this.fileSegmentTable;
-        return list.isEmpty() ? GET_FILE_SIZE_ERROR : list.get(list.size() - 1).getAppendOffset();
+        return list.isEmpty() ? 0L : list.get(list.size() - 1).getAppendOffset();
     }
 
     public long getMinTimestamp() {

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java

@@ -77,7 +77,7 @@ public void destroyExpiredFile(long expireTimestamp) {
         super.destroyExpiredFile(expireTimestamp);
         long afterOffset = this.getMinOffset();
 
-        if (beforeOffset != afterOffset) {
+        if (beforeOffset != afterOffset && afterOffset > 0) {
             log.info("CommitLog min cq offset reset, filePath={}, offset={}, expireTimestamp={}, change={}-{}",
                 filePath, firstOffset.get(), expireTimestamp, beforeOffset, afterOffset);
             firstOffset.set(GET_OFFSET_ERROR);

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatMessageFile.java

@@ -25,7 +25,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import org.apache.commons.lang3.StringUtils;
@@ -59,7 +58,6 @@ public class FlatMessageFile implements FlatFileInterface {
     protected final MetadataStore metadataStore;
     protected final FlatCommitLogFile commitLog;
     protected final FlatConsumeQueueFile consumeQueue;
-    protected final AtomicLong lastDestroyTime;
 
     protected final ConcurrentMap<String, CompletableFuture<?>> inFlightRequestMap;
 
@@ -77,7 +75,6 @@ public FlatMessageFile(FlatFileFactory fileFactory, String filePath) {
         this.metadataStore = fileFactory.getMetadataStore();
         this.commitLog = fileFactory.createFlatFileForCommitLog(filePath);
         this.consumeQueue = fileFactory.createFlatFileForConsumeQueue(filePath);
-        this.lastDestroyTime = new AtomicLong();
         this.inFlightRequestMap = new ConcurrentHashMap<>();
     }
 
@@ -388,8 +385,8 @@ public void shutdown() {
         closed = true;
         fileLock.lock();
         try {
-            commitLog.shutdown();
             consumeQueue.shutdown();
+            commitLog.shutdown();
         } finally {
             fileLock.unlock();
         }
@@ -399,8 +396,8 @@ public void shutdown() {
     public void destroyExpiredFile(long timestamp) {
         fileLock.lock();
         try {
-            commitLog.destroyExpiredFile(timestamp);
             consumeQueue.destroyExpiredFile(timestamp);
+            commitLog.destroyExpiredFile(timestamp);
         } finally {
             fileLock.unlock();
         }
@@ -410,8 +407,8 @@ public void destroy() {
         this.shutdown();
         fileLock.lock();
         try {
-            commitLog.destroyExpiredFile(Long.MAX_VALUE);
             consumeQueue.destroyExpiredFile(Long.MAX_VALUE);
+            commitLog.destroyExpiredFile(Long.MAX_VALUE);
             if (queueMetadata != null) {
                 metadataStore.deleteQueue(queueMetadata.getQueue());
             }