[ISSUE #9379] Fix timeStoreTable delete logic in IndexService (#9384)

* [ISSUE #9379] Fix timeStoreTable delete logic in IndexService

* [ISSUE #9379] Fix delete logic from TimeStoreTable in IndexService

Author: lizhimins

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

@@ -253,7 +253,7 @@ public void destroyExpiredFile(long expireTimestamp) {
                 FileSegment fileSegment = fileSegmentTable.get(0);
 
                 if (fileSegment.getMaxTimestamp() != Long.MAX_VALUE &&
-                    fileSegment.getMaxTimestamp() > expireTimestamp) {
+                    fileSegment.getMaxTimestamp() >= expireTimestamp) {
                     log.debug("FileSegment has not expired, filePath={}, fileType={}, " +
                             "offset={}, expireTimestamp={}, maxTimestamp={}", filePath, fileType,
                         fileSegment.getBaseOffset(), expireTimestamp, fileSegment.getMaxTimestamp());

tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreService.java

@@ -254,12 +254,12 @@ public CompletableFuture<List<IndexItem>> queryAsync(
                 .whenComplete((v, t) -> {
                     // Try to return the query results as much as possible here
                     // rather than directly throwing exceptions
-                    if (result.isEmpty() && t != null) {
-                        future.completeExceptionally(t);
-                    } else {
-                        List<IndexItem> resultList = new ArrayList<>(result.values());
-                        future.complete(resultList.subList(0, Math.min(resultList.size(), maxCount)));
+                    if (t != null) {
+                        log.error("IndexStoreService#queryAsync, topicId={}, key={}, maxCount={}, timestamp={}-{}",
+                            topic, key, maxCount, beginTime, endTime, t);
                     }
+                    List<IndexItem> resultList = new ArrayList<>(result.values());
+                    future.complete(resultList.subList(0, Math.min(resultList.size(), maxCount)));
                 });
         } catch (Exception e) {
             log.error("IndexStoreService#queryAsync, topicId={}, key={}, maxCount={}, timestamp={}-{}",
@@ -344,10 +344,15 @@ public void destroyExpiredFile(long expireTimestamp) {
         // delete file in time store table
         readWriteLock.writeLock().lock();
         try {
-            timeStoreTable.entrySet().removeIf(entry ->
-                entry.getKey() < expireTimestamp &&
-                    IndexFile.IndexStatusEnum.UPLOAD.equals(entry.getValue().getFileStatus()));
             flatAppendFile.destroyExpiredFile(expireTimestamp);
+            timeStoreTable.entrySet().removeIf(entry ->
+                IndexFile.IndexStatusEnum.UPLOAD.equals(entry.getValue().getFileStatus()) &&
+                    entry.getKey() < flatAppendFile.getMinTimestamp());
+            int tableSize = (int) timeStoreTable.entrySet().stream()
+                .filter(entry -> IndexFile.IndexStatusEnum.UPLOAD.equals(entry.getValue().getFileStatus()))
+                .count();
+            log.info("IndexStoreService delete file, timestamp={}, remote={}, table={}, all={}",
+                expireTimestamp, flatAppendFile.getFileSegmentList().size(), tableSize, timeStoreTable.size());
         } finally {
             readWriteLock.writeLock().unlock();
         }

tieredstore/src/test/java/org/apache/rocketmq/tieredstore/index/IndexStoreServiceTest.java

@@ -33,15 +33,18 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.apache.rocketmq.common.ThreadFactoryImpl;
 import org.apache.rocketmq.store.logfile.DefaultMappedFile;
 import org.apache.rocketmq.tieredstore.MessageStoreConfig;
 import org.apache.rocketmq.tieredstore.common.AppendResult;
+import org.apache.rocketmq.tieredstore.file.FlatAppendFile;
 import org.apache.rocketmq.tieredstore.file.FlatFileFactory;
 import org.apache.rocketmq.tieredstore.metadata.DefaultMetadataStore;
 import org.apache.rocketmq.tieredstore.metadata.MetadataStore;
 import org.apache.rocketmq.tieredstore.util.MessageStoreUtil;
 import org.apache.rocketmq.tieredstore.util.MessageStoreUtilTest;
+import org.awaitility.Awaitility;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -206,6 +209,39 @@ public void runServiceTest() throws InterruptedException {
         });
     }
 
+    @Test
+    public void deleteFileTest() throws InterruptedException, IllegalAccessException {
+        indexService = new IndexStoreService(fileAllocator, filePath);
+        indexService.start();
+
+        for (int i = 0; i < 2 * 20; i++) {
+            AppendResult result = indexService.putKey(
+                TOPIC_NAME, TOPIC_ID, QUEUE_ID, Collections.singleton(String.valueOf(i)),
+                i * 100L, MESSAGE_SIZE, System.currentTimeMillis());
+            Assert.assertEquals(AppendResult.SUCCESS, result);
+            TimeUnit.MILLISECONDS.sleep(1);
+        }
+
+        indexService.wakeup();
+        Awaitility.await().until(() -> {
+            int tableSize = (int) indexService.getTimeStoreTable().entrySet().stream()
+                .filter(entry -> IndexFile.IndexStatusEnum.UPLOAD.equals(entry.getValue().getFileStatus()))
+                .count();
+            return tableSize == 2;
+        });
+
+        long timestamp = indexService.getTimeStoreTable().firstEntry().getValue().getEndTimestamp();
+        FlatAppendFile flatAppendFile = (FlatAppendFile)
+            FieldUtils.readField(indexService, "flatAppendFile", true);
+
+        indexService.destroyExpiredFile(timestamp);
+        Assert.assertEquals(2, flatAppendFile.getFileSegmentList().size());
+        Assert.assertEquals(3, indexService.getTimeStoreTable().size());
+        indexService.destroyExpiredFile(timestamp + 1);
+        Assert.assertEquals(1, flatAppendFile.getFileSegmentList().size());
+        Assert.assertEquals(2, indexService.getTimeStoreTable().size());
+    }
+
     @Test
     public void restartServiceTest() throws InterruptedException {
         indexService = new IndexStoreService(fileAllocator, filePath);