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

[ISSUE #5542] Fix ConsumerProcessor lockBatchMQ future allOf data race issue

Browse Source
This commit is contained in:
zhouxiang
2022-11-21 14:28:08 +08:00
committed by Zhouxiang Zhan
parent d781271654
commit 991af5e204
1 changed files with 5 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/processor/ConsumerProcessor.java
+5 -15
View File
@@ -313,19 +313,15 @@ public class ConsumerProcessor extends AbstractProcessor {
Set<MessageQueue> successSet = new CopyOnWriteArraySet<>();
Set<AddressableMessageQueue> addressableMessageQueueSet = buildAddressableSet(mqSet);
Map<String, List<AddressableMessageQueue>> messageQueueSetMap = buildAddressableMapByBrokerName(addressableMessageQueueSet);
List<CompletableFuture<Set<MessageQueue>>> futureList = new ArrayList<>();
List<CompletableFuture<Void>> futureList = new ArrayList<>();
messageQueueSetMap.forEach((k, v) -> {
LockBatchRequestBody requestBody = new LockBatchRequestBody();
requestBody.setConsumerGroup(consumerGroup);
requestBody.setClientId(clientId);
requestBody.setMqSet(v.stream().map(AddressableMessageQueue::getMessageQueue).collect(Collectors.toSet()));
CompletableFuture<Set<MessageQueue>> future0 = new CompletableFuture<>();
try {
future0 = serviceManager.getMessageService().lockBatchMQ(ctx, v.get(0), requestBody, timeoutMillis);
future0.thenAccept(successSet::addAll);
} catch (Throwable t) {
future0.completeExceptionally(t);
}
CompletableFuture<Void> future0 = serviceManager.getMessageService()
.lockBatchMQ(ctx, v.get(0), requestBody, timeoutMillis)
.thenAccept(successSet::addAll);
futureList.add(FutureUtils.addExecutor(future0, this.executor));
});
CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).whenComplete((v, t) -> {
@@ -348,13 +344,7 @@ public class ConsumerProcessor extends AbstractProcessor {
requestBody.setConsumerGroup(consumerGroup);
requestBody.setClientId(clientId);
requestBody.setMqSet(v.stream().map(AddressableMessageQueue::getMessageQueue).collect(Collectors.toSet()));
CompletableFuture<Void> future0 = new CompletableFuture<>();
try {
future0 = serviceManager.getMessageService().unlockBatchMQ(ctx, v.get(0), requestBody, timeoutMillis);
future0.complete(null);
} catch (Throwable t) {
future0.completeExceptionally(t);
}
CompletableFuture<Void> future0 = serviceManager.getMessageService().unlockBatchMQ(ctx, v.get(0), requestBody, timeoutMillis);
futureList.add(FutureUtils.addExecutor(future0, this.executor));
});
CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0])).whenComplete((v, t) -> {