opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-19 02:23:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #7872] Fix Tiered Store Query Message Async return different view each time (#7874)

Browse Source 
Co-authored-by: ayue <ericyu0421@163.com>
This commit is contained in:
AYue
2024-03-26 15:12:38 +08:00
committed by GitHub
co-authored by ayue
parent dde4fcc3f6
commit d9c75ff1f9
1 changed files with 9 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tieredstore/src/main/java/org/apache/rocketmq/tieredstore/core/MessageStoreFetcherImpl.java
+9 -7
View File
@@ -394,8 +394,7 @@ public class MessageStoreFetcherImpl implements MessageStoreFetcher {
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 class MessageStoreFetcherImpl implements MessageStoreFetcher {
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, " +