JAVA / THREADS AND CONCURRENCY
CompletableFuture pipelines and async composition
Build non-blocking CompletableFuture chains with thenApply, thenCompose, thenCombine and allOf, and recover from failures without blocking mid-pipeline.
What you will learn
- Chain thenApply for plain transforms and thenCompose when a step returns a future
- Merge independent branches with thenCombine, and a whole list with allOf
- Recover with exceptionally or handle, unwrapping CompletionException via getCause()
- Keep join() at the pipeline boundary and never block inside a stage
Understanding CompletableFuture pipelines and async composition
The Future from the previous lesson is a read handle: once you hold it, the only way to reach the value is to block in get(). CompletableFuture inverts that by letting you attach the work that depends on a value before the value exists. A chain such as supplyAsync(...).thenApply(...).thenCompose(...) dedicates no thread to waiting; each stage is a callback that the thread which produced the previous result runs on its way out. The mental model is a wiring diagram rather than a script, so a single completion at the root pushes values through every dependent stage.
The three composition methods differ only in the shape of the function you hand them. thenApply takes T -> U and stores whatever the function returned as the stage's value, so if that function is itself an async call returning CompletableFuture<U>, you end up with CompletableFuture<CompletableFuture<U>> whose outer stage completes the moment the inner future object is created, not when the work is done. thenCompose takes T -> CompletionStage<U> and links the inner stage's completion to the outer one, which is the same map versus flatMap distinction you know from Optional and Stream. thenCombine is for two futures that do not depend on each other: it fires a BiFunction once both have values, which is how you keep two calls genuinely parallel instead of sequencing them.
A failure anywhere in the chain skips every downstream then* stage and travels to the first exceptionally, handle or whenComplete, arriving wrapped in a CompletionException, so test and log ex.getCause() rather than ex. Which thread runs a stage is also worth understanding: a non-async stage runs on whichever thread completed its predecessor, or on the calling thread if the value was already there, while the *Async variants hand the stage to the common ForkJoinPool or to an executor you supply. Because common-pool threads are daemons and the pool is sized for CPU work, blocking calls belong on your own executor, and you must join at the outer boundary or the JVM can exit with stages still pending.
import java.util.concurrent.CompletableFuture;
public class Pipeline {
// Stand-in for any async API: it hands back a future, not a value.
static CompletableFuture<Integer> lookupScore(String user) {
return CompletableFuture.supplyAsync(() -> user.length() * 10);
}
public static void main(String[] args) {
CompletableFuture<Integer> id = new CompletableFuture<>();
CompletableFuture<String> name = id
.thenApply(n -> "user-" + n) // Integer -> String
.thenApply(String::toUpperCase); // String -> String
CompletableFuture<Integer> score = name
.thenCompose(Pipeline::lookupScore); // String -> CF<Integer>, flattened
CompletableFuture<String> report = name
.thenCombine(score, (n, s) -> n + " scored " + s);
System.out.println("wired, report done? " + report.isDone());
id.complete(7); // the root value arrives; the whole graph runs from here
System.out.println(report.join());
System.out.println("score branch on its own: " + score.join());
}
}A CompletableFuture pipeline is a dependency graph you describe before any value exists, and the key decision at each link is whether the step returns a plain value (thenApply) or another future (thenCompose).
Worked examples
A failure travels down the chain
Shows that a thrown exception skips the remaining transforms and reaches exceptionally wrapped in CompletionException.
import java.util.concurrent.CompletableFuture;
public class Recover {
public static void main(String[] args) {
CompletableFuture<String> failed = CompletableFuture
.<String>supplyAsync(() -> { throw new IllegalStateException("no connection"); });
CompletableFuture<String> recovered = failed
.thenApply(v -> {
System.out.println("this stage is skipped");
return v.trim();
})
.exceptionally(ex -> "fallback: " + ex.getCause().getMessage()
+ " (caught as " + ex.getClass().getSimpleName() + ")");
System.out.println(recovered.join());
System.out.println("source failed? " + failed.isCompletedExceptionally());
}
}Example explained
Line 1The supplier throws, so that stage completes exceptionally and never holds a value.
Line 2thenApply never runs, because a stage fires only when its input completes normally.
Line 3exceptionally receives a CompletionException wrapper, so the IllegalStateException is at ex.getCause().
Line 4join() returns normally on the recovered stage even though the source stage stays failed.
Which thread runs a stage
Demonstrates that a non-async stage runs on the thread that completed its predecessor, or on the caller if the value is already present.
import java.util.concurrent.CompletableFuture;
public class WhoRuns {
public static void main(String[] args) throws InterruptedException {
CompletableFuture<String> source = new CompletableFuture<>();
CompletableFuture<Integer> length = source.thenApply(s -> {
System.out.println("thenApply runs on " + Thread.currentThread().getName());
return s.length();
});
Thread producer = new Thread(() -> source.complete("hello"), "producer");
producer.start();
producer.join();
System.out.println("length = " + length.join());
CompletableFuture.completedFuture("done").thenAccept(v ->
System.out.println("already-complete stage runs on "
+ Thread.currentThread().getName()));
}
}Example explained
Line 1The stage is registered while source is still pending, so nothing runs at registration time.
Line 2source.complete("hello") makes the producer thread push the value through the dependent stage before it returns.
Line 3thenAccept on an already-completed future runs the action inline on the caller, which here is main.
Line 4Neither thread is guaranteed by the API, which is why a stage must never assume it is on main.
Fan out and fan in with allOf
Runs three calls concurrently and gathers their results once allOf reports that every one of them finished.
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
public class FanIn {
static CompletableFuture<String> fetch(String host) {
return CompletableFuture.supplyAsync(() -> host + ":" + host.length());
}
public static void main(String[] args) {
List<CompletableFuture<String>> calls =
List.of(fetch("alpha"), fetch("bee"), fetch("gamma"));
String joined = CompletableFuture
.allOf(calls.toArray(new CompletableFuture[0]))
.thenApply(v -> calls.stream()
.map(CompletableFuture::join)
.collect(Collectors.joining(", ")))
.join();
System.out.println(joined);
}
}Example explained
Line 1All three futures are created before anything is joined, so the three suppliers overlap in the pool.
Line 2allOf returns CompletableFuture<Void>: it reports completion only, so v is null and the results must be re-read.
Line 3Inside that thenApply every future is already done, so each join() returns immediately and cannot block.
Line 4Streaming the original list preserves input order, which the completion order of the three calls does not.
Important notes
supplyAsync and every *Async call without an explicit executor use the shared ForkJoinPool.commonPool(), which is sized for CPU work; give blocking I/O its own executor or one slow call starves every other pipeline in the process.
join() throws an unchecked CompletionException while get() throws a checked ExecutionException; both wrap the original cause, and cancelling a CompletableFuture does not interrupt a supplier that is already running.
Common mistakes
Passing a future-returning method to thenApply instead of thenCompose: the result is CompletableFuture<CompletableFuture<X>>, and the outer stage reports done while the inner call is still in flight, so join() hands you an unfinished future and any inner failure goes unnoticed.
Calling get() or join() inside a thenApply body to wait for another future: that stage sits on a common-pool worker while it blocks, and once every worker is parked this way the remaining stages have no thread left to run on.
Building the pipeline and letting main return without joining anything: common-pool threads are daemons, so the JVM exits and the later stages simply never execute or print.
Try it yourself
Change, predict, then run
Given a CompletableFuture<Integer> id and a method CompletableFuture<String> loadName(int id), build a chain that yields the uppercase name, falls back to "unknown" on failure, and prints it with a single join(). Then swap thenCompose for thenApply and note the exact type the compiler reports.
Open the Java workspaceCheck your understanding
loadProfile(int id) returns CompletableFuture<Profile>. You write ids.thenApply(this::loadProfile). What do you actually get?
- A CompletableFuture<Profile>, because thenApply unwraps a returned future automatically
- A compile error, since the function passed to thenApply may not return a CompletableFuture
- A CompletableFuture<CompletableFuture<Profile>> that completes as soon as loadProfile hands back its future, before the profile has loaded
- A CompletableFuture<Profile> whose join() blocks the calling thread until loadProfile finishes
Show answer
thenApply treats whatever the function returns as an ordinary result, so the inner future becomes the stage's value and the outer stage is done before the profile arrives; thenCompose exists precisely to link the inner stage's completion to the outer one. Option 0 is tempting because both methods look like a map over the value, but flattening is the one thing thenApply does not do.