[ISSUE #8438] Fix broker return two messages when query message and i…

…ndex service bug
apache · Jul 24, 2024 · 200e7ed · 200e7ed
1 parent 6e6319f
commit 200e7ed
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 1 deletion.
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageStore.java
@@ -460,6 +460,9 @@ public synchronized void shutdown() {
         if (flatFileStore != null) {
             flatFileStore.shutdown();
         }
+        if (indexService != null) {
+            indexService.shutdown();
+        }
         if (storeExecutor != null) {
             storeExecutor.shutdown();
         }

diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/file/FlatAppendFile.java
@@ -90,6 +90,7 @@ public void recoverFileSize() {
     public void initOffset(long offset) {
         if (this.fileSegmentTable.isEmpty()) {
             FileSegment fileSegment = fileSegmentFactory.createSegment(fileType, filePath, offset);
+            fileSegment.initPosition(fileSegment.getSize());
             this.flushFileSegmentMeta(fileSegment);
             this.fileSegmentTable.add(fileSegment);
         }

diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreFile.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/index/IndexStoreFile.java
@@ -287,8 +287,12 @@ protected CompletableFuture<List<IndexItem>> queryAsyncFromUnsealedFile(
                     buffer.position(this.getItemPosition(slotValue));
                     buffer.get(bytes);
                     IndexItem indexItem = new IndexItem(bytes);
+                    long storeTimestamp = indexItem.getTimeDiff() + beginTimestamp.get();
                     if (hashCode == indexItem.getHashCode()) {
-                        result.add(indexItem);
+                        if (hashCode == indexItem.getHashCode() &&
+                            beginTime <= storeTimestamp && storeTimestamp <= endTime) {
+                            result.add(indexItem);
+                        }
                         if (result.size() > maxCount) {
                             break;
                         }