Fix race condition in consumer redelivery
@@ -1916,20 +1916,20 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (messageIds == null || messageIds.isEmpty()) {
return CompletableFuture.completedFuture(Collections.emptyList());
}
- List<MessageIdData> data = new ArrayList<>(messageIds.size());
- List<CompletableFuture<Void>> futures = new ArrayList<>(messageIds.size());
- messageIds.forEach(messageId -> {
+ List<CompletableFuture<MessageIdData>> futures = messageIds.stream().map(messageId -> {
CompletableFuture<Boolean> future = processPossibleToDLQ(messageId);
- futures.add(future.thenAccept(sendToDLQ -> {
+ return future.thenApply(sendToDLQ -> {
if (!sendToDLQ) {
- data.add(new MessageIdData()
+ return new MessageIdData()
.setPartition(messageId.getPartitionIndex())
.setLedgerId(messageId.getLedgerId())
- .setEntryId(messageId.getEntryId()));
+ .setEntryId(messageId.getEntryId());
}
- }));
+ return null;
});
- return FutureUtil.waitForAll(futures).thenCompose(v -> CompletableFuture.completedFuture(data));
+ }).collect(Collectors.toList());
+ return FutureUtil.waitForAll(futures).thenApply(v ->
+ futures.stream().map(CompletableFuture::join).filter(Objects::nonNull).collect(Collectors.toList()));
}
private CompletableFuture<Boolean> processPossibleToDLQ(MessageIdImpl messageId) {
