Found a problem in production environment regarding the CompletableFuture.supplyAsync()
We have a batch processing method like below:
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
public class CompletableFutureProblem {
public void batchOperation(){
List<String> stringList = new ArrayList<>();
stringList.add("task1");
stringList.add("task2");
List<CompletableFuture<String>> futures = new ArrayList<>();
stringList.parallelStream().forEach(str -> {
CompletableFuture<String> response = restApiCall(str);
futures.add(response);
});
//futures.add(null);
CompletableFuture<Void> result = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
CompletableFuture<List<String>> convertedResult = result.thenApply(v ->
futures.stream().map(CompletableFuture::join).collect(Collectors.toList())
);
try {
List<String> finishedTask = convertedResult.get();
System.out.println(finishedTask.toString());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
public CompletableFuture<String> restApiCall(String str){
return CompletableFuture.supplyAsync(() -> {
return "Complete-" + str;
});
}
public static void main(String[] args) {
CompletableFutureProblem problem = new CompletableFutureProblem();
problem.batchOperation();
}
}
When it works all right will print: [Complete-task2, Complete-task1]
However, sometimes it throws exception like below in production:
Exception in thread "main" java.lang.NullPointerException
at java.util.concurrent.CompletableFuture.andTree(CompletableFuture.java:1320)
at java.util.concurrent.CompletableFuture.allOf(CompletableFuture.java:2238)
at third.concurrent.CompletableFutureProblem.batchOperation(CompletableFutureProblem.java:20)
at third.concurrent.CompletableFutureProblem.main(CompletableFutureProblem.java:40)
I have investigated the CompletableFuture.allOf() source code found that if the list futures contains null, for example, futures.add(null), the exception will throw, but I really do not know under what scenarios will the CompletableFuture.supplyAsync() in restApiCall method return null ?
Thank you for your patient for reading the long post.
futures is being written to by multiple threads, since you are consuming stringList with a parallel stream. However futures is an ArrayList, which is not thread-safe.
Therefore, you can't be sure that each element added to it from a different thread will be visible without proper synchronization. When you transform it into an array, there will be memory visibility problems, which are undeterministic, hence why sometimes it works as expected.
To fix this problem, normally a concurrent collection would be used. However in this case, it does not make sense to parallelize CompletableFuture.supplyAsync(), since it is a non-blocking call. Therefore, the best solution is to just loop through the list:
stringList.forEach(str -> {
Also, the pre-allocated array in toArray() should be empty:
futures.toArray(new CompletableFuture[0])