[ISSUE apache#7872] Fix Tiered Store Query Message Async return diffe…

…rent view each time (apache#7874) Co-authored-by: ayue <[email protected]>
Jianing-Xu · Mar 26, 2024 · d9c75ff · d9c75ff
1 parent dde4fcc
commit d9c75ff
Showing 1 changed file with 9 additions and 7 deletions.
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java
@@ -394,8 +394,7 @@ public CompletableFuture<QueryMessageResult> queryMessageAsync(
             messageStore.getIndexService().queryAsync(topic, key, maxCount, begin, end);
 
         return future.thenCompose(indexItemList -> {
-            QueryMessageResult result = new QueryMessageResult();
-            List<CompletableFuture<Void>> futureList = new ArrayList<>(maxCount);
+            List<CompletableFuture<SelectMappedBufferResult>> futureList = new ArrayList<>(maxCount);
             for (IndexItem indexItem : indexItemList) {
                 if (topicId != indexItem.getTopicId()) {
                     continue;
@@ -405,17 +404,20 @@ public CompletableFuture<QueryMessageResult> queryMessageAsync(
                 if (flatFile == null) {
                     continue;
                 }
-                CompletableFuture<Void> getMessageFuture = flatFile
+                CompletableFuture<SelectMappedBufferResult> getMessageFuture = flatFile
                     .getCommitLogAsync(indexItem.getOffset(), indexItem.getSize())
-                    .thenAccept(messageBuffer -> result.addMessage(
-                        new SelectMappedBufferResult(
-                            indexItem.getOffset(), messageBuffer, indexItem.getSize(), null)));
+                    .thenApply(messageBuffer -> new SelectMappedBufferResult(
+                        indexItem.getOffset(), messageBuffer, indexItem.getSize(), null));
                 futureList.add(getMessageFuture);
                 if (futureList.size() >= maxCount) {
                     break;
                 }
             }
-            return CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).thenApply(v -> result);
+            return CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).thenApply(v -> {
+                QueryMessageResult result = new QueryMessageResult();
+                futureList.forEach(f -> f.thenAccept(result::addMessage));
+                return result;
+            });
         }).whenComplete((result, throwable) -> {
             if (result != null) {
                 log.info("MessageFetcher#queryMessageAsync, " +