Correct use of CompletableFuture.anyOf()

Question

I want to make parallel calls to different external services and proceed with the first successful response.

Successful here means that the result returned by the service has certain values in certain fields.

However, for CompletableFuture, everything other than an exception is success. So, even for for business failures, I have to throw an exception to signal non-success to CompletableFuture. This feels wrong, ideally I would want to provide a boolean to indicate business success/failure. Is there a better way to signal business failures?

The second question I have is, how do I make sure I don't run out of threads due to the abandoned CompletableFutures that would keep running even after CompletableFutures.anyOf() returns. Ideally I want to force stop the threads but as per the thread below, the best I can do is cancel the downstream operations.

How to cancel Java 8 completable future?

When you call CompletableFuture#cancel, you only stop the downstream part of the chain. Upstream part, i. e. something that will eventually call complete(...) or completeExceptionally(...), doesn't get any signal that the result is no more needed.

I can force stop treads by providing my own ExecutorService and calling shutdown()/shutdownNow() on it after CompletableFuture.anyOf() returns. However I am not sure about the implications of creating a new ExecutorService instance for each request.

Answer 1

This is perhaps an attempt to look away from anyOf(), as there's just no easy way to cancel other tasks with it.

What this is doing is create and start async tasks, then keep a reference to the future along with an object that would be used to effectively terminate other tasks, which of course is dependent on your actual code/task. In this example, I'm just returning the input string (hence the type Pair<CompletableFuture<String>, String>); your code will probably have something like a request object.

ExecutorService exec = Executors.newFixedThreadPool(5);
List<String> source = List.of();

List<Pair<CompletableFuture<String>, String>> futures = source.stream()
        .map(s -> Pair.of(CompletableFuture.supplyAsync(() -> s.toUpperCase(), exec), s))
        .collect(Collectors.toList());

Consumer<Pair<CompletableFuture<String>, String>> cancelOtherTasksFunction = pair -> {
    futures.stream()
            .filter(future -> future != pair)
            .forEach(future -> {
                future.getLeft().cancel(true); // cancel the future
                // future.getRight().doSomething() // cancel actual task
            });
};

AtomicReference<String> result = new AtomicReference<>();
futures.forEach(future -> future.getLeft()
        .thenAccept(s -> {
            if (null == result.compareAndExchangeRelease(null, s)) {
                // the first success is recorded
                cancelOtherTasksFunction.accept(future);
            }
        }));

I suppose you can (should?) create a class to hold the future and the task object (such as an http request you could cancel) to replace the Pair.

Note:

• if (null == result.compareAndExchangeRelease(null, s)) is only valid if your future returns a non-null value.
• The first valid response will be in result, but you still need to test for blocking before returning it, although I suppose it should work as other tasks are being cancelled (that's the theory).
• You may decide to make futures.forEach part of the stream pipeline above, just be careful to force all tasks to be submitted (which collect(Collectors.toList()) does).