[ISSUE apache#8076] Fix correct min cq offset when delete tiered stor…

…age CommitLog (apache#8082)
Jianing-Xu · May 7, 2024 · 0f0324a · 0f0324a
1 parent a15088c
commit 0f0324a
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 3 deletions.
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFile.java
@@ -60,4 +60,17 @@ public CompletableFuture<Long> getMinOffsetFromFileAsync() {
                 return firstOffset.get();
             });
     }
+
+    @Override
+    public void destroyExpiredFile(long expireTimestamp) {
+        long beforeOffset = this.getMinOffset();
+        super.destroyExpiredFile(expireTimestamp);
+        long afterOffset = this.getMinOffset();
+
+        if (beforeOffset != afterOffset) {
+            log.info("CommitLog min cq offset reset, filePath={}, offset={}, expireTimestamp={}, change={}-{}",
+                filePath, firstOffset.get(), expireTimestamp, beforeOffset, afterOffset);
+            firstOffset.set(GET_OFFSET_ERROR);
+        }
+    }
 }
diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFileTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatCommitLogFileTest.java
@@ -93,19 +93,33 @@ public void getMinOffsetFromFileAsyncTest() {
         for (int i = 6; i < 9; i++) {
             ByteBuffer byteBuffer = MessageFormatUtilTest.buildMockedMessageBuffer();
             byteBuffer.putLong(MessageFormatUtil.QUEUE_OFFSET_POSITION, i);
-            Assert.assertEquals(AppendResult.SUCCESS, flatFile.append(byteBuffer, 1L));
+            Assert.assertEquals(AppendResult.SUCCESS, flatFile.append(byteBuffer, i));
         }
         Assert.assertEquals(-1L, flatFile.getMinOffsetFromFileAsync().join().longValue());
 
         // append some messages
         for (int i = 9; i < 30; i++) {
+            if (i == 20) {
+                flatFile.commitAsync().join();
+                flatFile.rollingNewFile(flatFile.getAppendOffset());
+            }
             ByteBuffer byteBuffer = MessageFormatUtilTest.buildMockedMessageBuffer();
             byteBuffer.putLong(MessageFormatUtil.QUEUE_OFFSET_POSITION, i);
-            Assert.assertEquals(AppendResult.SUCCESS, flatFile.append(byteBuffer, 1L));
+            Assert.assertEquals(AppendResult.SUCCESS, flatFile.append(byteBuffer, i));
         }
 
         flatFile.commitAsync().join();
         Assert.assertEquals(6L, flatFile.getMinOffsetFromFile());
         Assert.assertEquals(6L, flatFile.getMinOffsetFromFileAsync().join().longValue());
+
+        // recalculate min offset here
+        flatFile.destroyExpiredFile(20L);
+        Assert.assertEquals(20L, flatFile.getMinOffsetFromFile());
+        Assert.assertEquals(20L, flatFile.getMinOffsetFromFileAsync().join().longValue());
+
+        // clean expired file again
+        flatFile.destroyExpiredFile(20L);
+        Assert.assertEquals(20L, flatFile.getMinOffsetFromFile());
+        Assert.assertEquals(20L, flatFile.getMinOffsetFromFileAsync().join().longValue());
     }
 }
diff --git a/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatFileStoreTest.java b/tieredstore/src/test/java/org/apache/rocketmq/tieredstore/file/FlatFileStoreTest.java
@@ -46,7 +46,7 @@ public void init() {
         storeConfig = new MessageStoreConfig();
         storeConfig.setStorePathRootDir(storePath);
         storeConfig.setTieredBackendServiceProvider(PosixFileSegment.class.getName());
-        storeConfig.setBrokerName(storeConfig.getBrokerName());
+        storeConfig.setBrokerName("brokerName");
         metadataStore = new DefaultMetadataStore(storeConfig);
     }