CompletableFuture return all finished result after timeout

Question

I want to have a list of comletablefutures I want to wait. With following code.

  public static <T> CompletableFuture<List<T>> finishAllQuery(
  List<CompletableFuture<T>> futures) {
CompletableFuture<Void> allDoneFuture =
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]));
return allDoneFuture.thenApply(
    v -> futures.stream().filter(Objects::nonNull)
        .map(future -> future.join())
        .collect(Collectors.toList())
);
  }
CompletableFuture<List<Result>> allResponse = finishAllQuery(futures);
allResponse.get(5, milliseconds)

The problem is that among all the futures, some of them can be slow, I want that after the expiration time, the get method return with all completed results. Is there a way to do that?

Thanks a lot.

Answer 1

This should be handled by finishAllQuery itself. E.g., starting with Java 9, you can use

public static <T> CompletableFuture<List<T>> finishAllQuery(
    List<CompletableFuture<T>> futures, long timeOut, TimeUnit unit) {

    return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
        .completeOnTimeout(null, timeOut, unit)
        .thenApply(v -> futures.stream()
            .filter(CompletableFuture::isDone)
            .map(CompletableFuture::join)
            .collect(Collectors.toList()));
}

With completeOnTimeout, we can force the completion of the future with a predefined value after the timeout elapsed. Here, we just use null, as the result value of allOf doesn’t matter anyway.

We just have to add a filter condition, to skip all futures which are not completed yet, as otherwise, join would block the thread.

This can be used like

CompletableFuture<List<Result>> allResponse
    = finishAllQuery(futures, 5, TimeUnit.MILLISECONDS);
List<Result> list = allResponse.join(); // will wait at most 5 milliseconds

For Java 8, we can use

static <T> CompletableFuture<T> completeOnTimeout(
    CompletableFuture<T> cf, T value, long timeOut, TimeUnit unit) {

    ScheduledExecutorService e = Executors.newSingleThreadScheduledExecutor();
    ScheduledFuture<Boolean> job = e.schedule(() -> cf.complete(value), timeOut, unit);
    return cf.whenComplete((x,y) -> { job.cancel(false); e.shutdown(); });
}

for the missing feature, which requires a small rewrite:

public static <T> CompletableFuture<List<T>> finishAllQuery(
    List<CompletableFuture<T>> futures, long timeOut, TimeUnit unit) {
    return completeOnTimeout(
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])),
        null, timeOut, unit)
        .thenApply(v -> futures.stream()
            .filter(CompletableFuture::isDone)
            .map(CompletableFuture::join)
            .collect(Collectors.toList()));
}

The way, the caller uses the method, doesn’t change.

For production use, it’s worth rewriting the completeOnTimeout method to reuse the ScheduledExecutorService, but this also requires adding shutdown code or a thread factory creating daemon threads. With Java 9 or newer, you get that for free.