1. Overview
In this tutorial, we’ll learn how to convert a List<CompletableFuture<T>> object to a CompletableFuture<List<T>>.
This conversion can be very useful in many cases. A prime example would be when we have to make multiple calls to a remote service, typically an asynchronous operation, and aggregate the results into a single List. Additionally, we end up waiting on a single CompletableFuture object which provides us the results list when all operations are finished or throws an exception if one or more end in failure.
We’ll first see a naïve way of doing the conversion and then look at a simpler and safer approach.
2. Chaining CompletableFutures
One way of doing this is to chain the CompletableFutures using their thenCompose() method. This way, we can create a single object that resolves once all the previous futures resolve, one by one, akin to a domino construct.
2.1. Implementation
First, let’s create a mock asynchronous operation:
public class Application {
ScheduledExecutorService asyncOperationEmulation;
Application initialize() {
asyncOperationEmulation = Executors.newScheduledThreadPool(10);
return this;
}
CompletableFuture<String> asyncOperation(String operationId) {
CompletableFuture<String> cf = new CompletableFuture<>();
asyncOperationEmulation.submit(() -> {
Thread.sleep(100);
cf.complete(operationId);
});
return cf;
}
We’ve created an Application class to host our test code and the asyncOperation() method which simply sleeps for 100 ms. We employ an Executor with 10 threads to dispatch everything asynchronously.
To gather all of our operation results, in this case, simple operationId strings, we’ll chain the CompletableFutures generated from the asyncOperation() method:
void startNaive() {
List<CompletableFuture<String>> futures = new ArrayList<>();
for (int i = 1; i <= 1000; i++) {
String operationId = "Naive-Operation-" + i;
futures.add(asyncOperation(operationId));
}
CompletableFuture<List<String>> aggregate = CompletableFuture.completedFuture(new ArrayList<>());
for (CompletableFuture<String> future : futures) {
aggregate = aggregate.thenCompose(list -> {
list.add(future.get());
return CompletableFuture.completedFuture(list);
});
}
final List<String> results = aggregate.join();
for (int i = 0; i < 10; i++) {
System.out.println("Finished " + results.get(i));
}
close();
}
We start by creating a completed CompleteableFuture using the static completedFuture() method and provide an empty List as the completion result. Using thenCompose() we create a Runnable that executes once the previous future has finished, in this case immediately. The thenCompose() method returns a new CompletableFuture which resolves once both the first and second future finish. We replace the aggregate reference with this new future object. This allows us to keep chaining these calls inside the iteration loop over the futures list.
Inside the Runnable we’ve created, we wait for the future to finish and add the result to the list. We then return a completed future with that list and the result. This will pass the list further down the thenCompose() chain, letting us add the future results one by one.
Once all futures are chained, we call join() on the aggregate CompletableFuture. This is done specifically for the example, so that we can retrieve the results and block the main Java thread from exiting before aggregate is finished. In a real asynchronous scenario we’d probably add our callback logic inside a thenAccept() or whenComplete() call.
One thing to notice is we add a close() call at the end with the following implementation:
void close() {
asyncOperationEmulation.shutdownNow();
}
Closing all Executors is mandatory when an application exits, otherwise, the Java process will hang.
2.2. Implementation Problems
The naïve implementation has a few problems. Not only the future chaining introduces unwanted complexity, but it also creates a large number of unneeded objects, such as all the new CompletableFutures generated by thenCompose().
Another potential issue appears when we execute a large number of operations. If an operation fails, and depending on how the Java implementation resolves the CompletableFuture chain, we might get a StackOverflowError if the resolutions are done recursively.
To test the exception scenario we can introduce an error on one of the operations by changing the asyncOperation() method:
CompletableFuture<String> asyncOperation(String operationId) {
CompletableFuture<String> cf = new CompletableFuture<>();
asyncOperationEmulation.submit(() -> {
if (operationId.endsWith("567")) {
cf.completeExceptionally(new Exception("Error on operation " + operationId));
return;
}
Thread.sleep(100);
cf.complete(operationId);
});
return cf;
}
The future for the 567th operation will complete exceptionally in this case, making the aggregate.join() call also throw a runtime exception.
3. Using CompletableFuture.allOf()
A different and better approach is to use the allOf() method of the CompletableFuture API. This method takes an array of CompletableFuture objects and creates a new one that resolves when all the provided futures themselves resolve.
Additionally, if one of the futures fails then the aggregate future also fails. The new future doesn’t contain the list of results. To obtain them we have to inspect the respective CompletableFuture object.
3.1. Implementation
Let’s create a new start() method using allOf():
void start() {
List<CompletableFuture<String>> futures = new ArrayList<>();
for (int i = 1; i <= 1000; i++) {
String operationId = "Operation-" + i;
futures.add(asyncOperation(operationId));
}
CompletableFuture<?>[] futuresArray = futures.toArray(new CompletableFuture<?>[0]);
CompletableFuture<List<String>> listFuture = CompletableFuture.allOf(futuresArray)
.thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()));
final List<String> results = listFuture.join();
System.out.println("Printing first 10 results");
for (int i = 0; i < 10; i++) {
System.out.println("Finished " + results.get(i));
}
close();
}
The setup and result printing are the same, however we now have a futuresArray and provide it to allOf(). We use thenApply() to add logic after allOf() is resolved. In this callback, we gather all futures results using the CompletableFuture.join() method and collect them into a List. This list is the result contained inside the CompletableFuture generated by thenApply(), namely the listFuture.
To showcase the aggregate results we use the join() method which blocks the main thread until listFuture is complete. We shouldn’t forget the close() call at the end.
3.2. Pros of allOf()
The allOf() based implementation is a simpler and cleaner way of handling multiple asynchronous operations than future chaining. The aggregate CompletableFuture provides atomicity to the whole operation and completes when all futures succeed or fail when even one fails. This protects us from potential partial processing of results.
Additionally, it lets us wait for all futures to complete in a non-blocking manner. Notice that in the example code, we call join() for the listFuture object but in a realistic scenario we’d rely on just the callback.
4. Conclusion
In this article, we learned how to convert a List<CompletableFuture<T>> into a CompletableFuture<List<T>>. We understood why this conversion is useful and saw two ways of doing it, one naïve implementation and one using the proper Java APIs. We discussed the potential issues with the former and how the latter avoids them.
As always, the source code for this article is available over on GitHub.
