Java 8 CompletableFuture: Combining and Error Handling

Introduction

This tutorial is the second part of the Java 8 CompletableFuture series. It explains how to combine several CompletableFutures and how to handle errors in a pipeline. We will start with the three ways of putting futures together - thenCompose() for dependent calls, thenCombine() for independent ones, and allOf() / anyOf() for any number of futures. We will then see how an exception travels through a pipeline, and how exceptionally(), handle() and whenComplete() deal with it. Lastly, we will run one example that shows all of these, and list the common mistakes.

(This tutorial assumes you know the basics covered in the Java 8 CompletableFuture tutorialClick to read tutorial: Java 8 CompletableFuture Tutorial with Examples: supplyAsync(), thenApply(), thenAccept() and join().)

Combining futures - three shapes

Real tasks rarely consist of one asynchronous call. You may need an employee's id before you can fetch their salary, or a salary and a bonus rate that can be fetched at the same time, or the salaries of a whole team. Each situation has its own method.
Java 8 CompletableFuture thenCompose, thenCombine, allOf and anyOf - dependent, independent and many futures

thenCompose() - one call after another

Definition of thenCompose() in CompletableFuture<T> is -
<U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)
Where,
  • fn receives the result of this future and returns another future, for example id -> fetchSalary(id).
  • The returned CompletableFuture<U> completes when that inner future completes. Using thenApply() with the same function would give a nested CompletableFuture<CompletableFuture<U>> instead. The relationship is the same as between map() and flatMap() on a streamClick to read: stream.

thenCombine() - two independent calls

Definition of thenCombine() in CompletableFuture<T> is -
<U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
Where,

allOf() and anyOf() - any number of futures

  • static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs) completes when all the given futures have completed. It returns Void, because the futures may all have different types; you collect the results by calling join() on each one afterwards, which no longer blocks.
  • static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) completes as soon as any one of them completes, with that future's result. It is useful for querying two mirrors of the same service and using whichever answers first.

How errors travel through a pipeline

If a stage throws an exception, the future it returns completes exceptionally. Every later stage that needs a value - thenApply(), thenCompose(), thenAccept() and the rest - is skipped, and its own future completes exceptionally with the same failure, wrapped in a CompletionException. The failure travels down the pipeline until it reaches a stage that handles exceptions, or until someone calls join() or get().
Java 8 CompletableFuture error handling - an exception skips later stages until exceptionally, handle or whenComplete
  • exceptionally(Function<Throwable, ? extends T> fn): called only if there was a failure, and returns a replacement value of the same type - the asynchronous version of a catch block.
  • handle(BiFunction<? super T, Throwable, ? extends U> fn): always called, with the value (and a null exception) on success, or a null value and the exception on failure. It returns a new value, possibly of a different type.
  • whenComplete(BiConsumer<? super T, ? super Throwable> action): always called with the same two arguments, but cannot change the outcome. The value or the failure is passed on unchanged - the asynchronous version of finally, typically used for logging.
  • completeExceptionally(Throwable ex): the manual counterpart of complete(), for failing a future you created yourself.

Java 8 code example of combining futures and handling errors

The example has three slow lookups - an employee id by name, a salary by id, and a bonus rate by department - each returning a CompletableFuture. findEmployeeId() throws an IllegalArgumentException for an unknown name, which parts 4 to 7 use to show error handling.
Java 8 code showing thenCompose, thenCombine, allOf, exceptionally, handle and whenComplete
//CombiningFutures.java
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.stream.Collectors;

public class CombiningFutures {

  static final Map<String, String> IDS = new HashMap<>();
  static final Map<String, Double> SALARIES = new HashMap<>();
  static {
    IDS.put("Tom Jones", "E-1001");
    IDS.put("Harry Major", "E-1002");
    IDS.put("Ethan Hardy", "E-1003");
    IDS.put("Nancy Smith", "E-1004");
    SALARIES.put("E-1001", 5200.0);
    SALARIES.put("E-1002", 4100.0);
    SALARIES.put("E-1003", 6300.0);
    SALARIES.put("E-1004", 5800.0);
  }

  // Three slow, asynchronous lookups
  static CompletableFuture<String> findEmployeeId(String name) {
    return CompletableFuture.supplyAsync(() -> {
      sleep(100);
      String id = IDS.get(name);
      if (id == null) {
        throw new IllegalArgumentException("No employee named " + name);
      }
      return id;
    });
  }

  static CompletableFuture<Double> fetchSalary(String id) {
    return CompletableFuture.supplyAsync(() -> {
      sleep(150);
      return SALARIES.get(id);
    });
  }

  static CompletableFuture<Double> fetchBonusRate(String department) {
    return CompletableFuture.supplyAsync(() -> {
      sleep(150);
      return 0.10;                       // the same rate for every department here
    });
  }

  static void sleep(long millis) {
    try {
      Thread.sleep(millis);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  public static void main(String[] args) {
    // 1. thenCompose: the second call needs the result of the first
    CompletableFuture<Double> salary =
        findEmployeeId("Tom Jones").thenCompose(id -> fetchSalary(id));
    System.out.println("1. thenCompose  -> salary " + salary.join());

    // 2. thenCombine: two independent calls, both results needed
    CompletableFuture<Double> bonus = fetchSalary("E-1001")
        .thenCombine(fetchBonusRate("Finance"), (pay, rate) -> pay * rate);
    System.out.println("2. thenCombine  -> bonus " + bonus.join());

    // 3. allOf: start many calls, wait for all, collect the results
    List<String> team =
        Arrays.asList("Tom Jones", "Harry Major", "Ethan Hardy", "Nancy Smith");
    List<CompletableFuture<Double>> calls = team.stream()
        .map(name -> findEmployeeId(name).thenCompose(CombiningFutures::fetchSalary))
        .collect(Collectors.toList());
    CompletableFuture<Void> all =
        CompletableFuture.allOf(calls.toArray(new CompletableFuture<?>[0]));
    List<Double> salaries = all
        .thenApply(v -> calls.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList()))
        .join();
    System.out.println("3. allOf        -> " + salaries);

    // 4. A failure skips the later stages; exceptionally() supplies a default
    double unknown = findEmployeeId("Deborah Sprightly")
        .thenCompose(CombiningFutures::fetchSalary)
        .thenApply(pay -> pay * 12)
        .exceptionally(ex -> {
          System.out.println("4. exceptionally caught " + ex.getClass().getSimpleName()
              + " caused by: " + ex.getCause().getMessage());
          return 0.0;
        })
        .join();
    System.out.println("   annual salary used: " + unknown);

    // 5. handle(): one method for both outcomes
    String report = findEmployeeId("Deborah Sprightly")
        .handle((id, ex) -> ex == null ? "found " + id : "not found")
        .join();
    System.out.println("5. handle       -> " + report);

    // 6. whenComplete(): look at the outcome without changing it
    CompletableFuture<String> logged = findEmployeeId("Harry Major")
        .whenComplete((id, ex) -> System.out.println("6. whenComplete saw " + id));
    System.out.println("   result still: " + logged.join());

    // 7. join() rethrows the failure, wrapped in an unchecked CompletionException
    try {
      findEmployeeId("Deborah Sprightly").join();
    } catch (CompletionException e) {
      System.out.println("7. join threw CompletionException, cause: "
          + e.getCause().getClass().getSimpleName());
    }
  }
}
 OUTPUT of the above code
1. thenCompose  -> salary 5200.0
2. thenCombine  -> bonus 520.0
3. allOf        -> [5200.0, 4100.0, 6300.0, 5800.0]
4. exceptionally caught CompletionException caused by: No employee named Deborah Sprightly
   annual salary used: 0.0
5. handle       -> not found
6. whenComplete saw E-1002
   result still: E-1002
7. join threw CompletionException, cause: IllegalArgumentException

Explanation of the code

  • Part 1: thenCompose() waits for the id and then starts fetchSalary(id), giving a flat CompletableFuture<Double>.
  • Part 2: fetchSalary() and fetchBonusRate() are both started before thenCombine() is called, so they run at the same time. The BiFunction multiplies the two results.
  • Part 3: a stream starts one id-then-salary pipeline per employee, all running at once. allOf() needs an array, hence toArray(new CompletableFuture<?>[0]). Once it completes, every future is done, so join() inside thenApply() returns immediately, and the salaries come back in the order of the list.
  • Part 4: the unknown name makes the first stage fail, so thenCompose() and thenApply() never run. exceptionally() receives a CompletionException whose cause is the original IllegalArgumentException, and returns 0.0 as the default. Use getCause() to reach the real exception.
  • Part 5: handle() turns both outcomes into a String; exactly one of id and ex is non-null.
  • Part 6: whenComplete() sees the result and prints it, and the future it returns completes with the same id.
  • Part 7: without any handling stage, join() throws the failure as an unchecked CompletionException. get() would throw a checked ExecutionException instead.

Common mistakes

  • Using thenApply() where thenCompose() is needed, ending up with a CompletableFuture<CompletableFuture<T>>.
  • Starting "parallel" calls one after the other: calling join() on the first future before creating the second makes them sequential. Create all the futures first, then combine them.
  • Swallowing failures: a pipeline that nobody joins and that has no exceptionally() or handle() stage loses its exceptions silently.
  • Forgetting the wrapper: inside exceptionally() and handle(), the exception is usually a CompletionException; check getCause() before deciding what went wrong.
  • No timeout: Java 8 has no orTimeout() (it was added in Java 9), so use get(timeout, unit) when a call may hang.

Summary

In this tutorial we first saw the three ways to put futures together: thenCompose() for a call that needs the previous result, thenCombine() for two independent calls, and allOf() and anyOf() for any number of futures. Next we saw how an exception skips the remaining stages, wrapped in a CompletionException, and how exceptionally(), handle() and whenComplete() differ. We then ran an example covering all of them. Lastly, we listed the common mistakes, from nesting futures with thenApply() to missing timeouts.