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

…rent view each time
AYue-94 · Feb 29, 2024 · 557e33e · 557e33e
1 parent eed303d
commit 557e33e
Showing 1 changed file with 8 additions and 7 deletions.
diff --git a/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java b/tieredstore/src/main/java/org/apache/rocketmq/tieredstore/TieredMessageFetcher.java
@@ -552,8 +552,7 @@ public CompletableFuture<QueryMessageResult> queryMessageAsync(
         CompletableFuture<List<IndexItem>> future = indexStoreService.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;
@@ -563,17 +562,19 @@ 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) {
                 LOGGER.info("MessageFetcher#queryMessageAsync, " +