05 · Concurrency Primer — The Functional Async Primitive

You have spent this phase making synchronous code readable. Concurrency is where that discipline pays or breaks: a badly composed CompletableFuture chain becomes a callback-hell relic, but a well-composed one reads like a stream pipeline for async values. This file is a bridge, not a full concurrency course — Phase 05 handles the JMM, virtual threads, and structured concurrency in depth. Your goal here is to learn CompletableFuture (CF) as a functional type that composes, and to know when Java 21 virtual threads make CF chains obsolete.

Why CompletableFuture Exists

Future<T> (Java 5) let you get() a value later — but get() blocks, and you cannot chain, combine, or handle errors without blocking a thread per operation. CompletableFuture<T> (Java 8) fixes this by exposing a monadic API: every operation returns a new CF, so you describe the pipeline once and the runtime executes it on a thread pool.

Mental model: CF is Optional for time. Both wrap a value that may or may not be available; both give you map-style operations to transform without unwrapping.

Optional<T>	CompletableFuture<T>	Meaning
map	thenApply	Transform value if present
flatMap	thenCompose	Chain another wrapped op
orElse	exceptionally	Fallback if empty/failed
ifPresent	thenAccept	Side-effect on value

Creating a CompletableFuture

Three sources cover 95% of real code:

// 1. Wrap a synchronous value (already complete)
CompletableFuture<String> done = CompletableFuture.completedFuture("ready");

// 2. Run async, no return (Runnable)
CompletableFuture<Void> fire = CompletableFuture.runAsync(() -> log("started"));

// 3. Compute async, returns a value (Supplier) — the workhorse
CompletableFuture<User> user = CompletableFuture.supplyAsync(() -> userRepo.find(id));

By default, runAsync/supplyAsync submit work to the common ForkJoinPool. For I/O-heavy tasks pass your own executor — the common pool is CPU-sized and blocking I/O starves it.

ExecutorService io = Executors.newFixedThreadPool(32);
CompletableFuture<User> user = CompletableFuture.supplyAsync(() -> userRepo.find(id), io);

The Three Composition Operators You Actually Use

Nearly every real chain is built from thenApply, thenCompose, and thenCombine. Learn these three cold; the other 40 methods are variations.

thenApply — Transform (Function<T, R>)

Same async value, new shape. Do NOT do async work inside — this runs on the completing thread.

CompletableFuture<String> name =
    CompletableFuture.supplyAsync(() -> userRepo.find(id))  // CF<User>
                     .thenApply(User::name);                // CF<String>

thenCompose — Flat-map (Function<T, CF>)

The one that saves you from CompletableFuture<CompletableFuture<Order>>. Use whenever the next step is itself async.

CompletableFuture<Order> order =
    CompletableFuture.supplyAsync(() -> userRepo.find(userId))
                     .thenCompose(user ->
                         CompletableFuture.supplyAsync(() -> orderRepo.latestFor(user)));

If you had used thenApply here, you would get CF<CF<Order>> — the exact same trap as Optional<Optional<T>> from the last file.

thenCombine — Zip two independent CFs (BiFunction<T, U, R>)

Run two async ops in parallel, join their results.

CompletableFuture<User>  userCf  = CompletableFuture.supplyAsync(() -> userRepo.find(id), io);
CompletableFuture<Cart>  cartCf  = CompletableFuture.supplyAsync(() -> cartRepo.find(id), io);

CompletableFuture<Checkout> checkout =
    userCf.thenCombine(cartCf, (u, c) -> new Checkout(u, c));

Both fire simultaneously; the BiFunction runs when both complete. Beats sequential thenCompose by the latency of the slower call.

⚠️ What Most People Get Wrong

They call .get() on a CompletableFuture inside another lambda. That blocks a pooled thread and defeats the whole point. Chain with thenCompose/thenCombine; only .get() (or better, .join()) at the very edge of your program — the servlet response, the CLI’s main thread, the test assertion. If you find yourself blocking in the middle of a pipeline, refactor.

Exception Handling — exceptionally, handle, whenComplete

Exceptions inside a CF chain do not bubble like sync code — they wrap into a CompletionException and short-circuit downstream steps. You need explicit recovery.

Operator	Signature	Use when
exceptionally	Function<Throwable, T>	Fallback value on failure, ignore success
handle	BiFunction<T, Throwable, R>	Transform both outcomes into one shape
whenComplete	BiConsumer<T, Throwable>	Side-effect (log/metrics), no transform

CompletableFuture<Price> price =
    priceService.quoteAsync(sku)
        .exceptionally(ex -> {
            log.warn("quote failed for {}", sku, ex);
            return Price.unavailable();          // fallback
        });

CompletableFuture<PriceResult> result =
    priceService.quoteAsync(sku)
        .handle((p, ex) -> ex == null
            ? PriceResult.ok(p)
            : PriceResult.error(ex.getMessage()));  // unify outcomes

priceService.quoteAsync(sku)
        .whenComplete((p, ex) -> metrics.record("quote", ex == null));  // observe only

Prefer handle over exceptionally when the success and failure types differ — it forces you to think about both branches. If you already have a sealed Result<T, E> from Phase 03, wrap here and the rest of your code stays typed.

A Realistic Chain

Fetch user + cart in parallel, then price the cart, then apply a discount — with a fallback if pricing fails:

public CompletableFuture<Checkout> buildCheckout(long userId) {
    var userCf = CompletableFuture.supplyAsync(() -> userRepo.find(userId), io);
    var cartCf = CompletableFuture.supplyAsync(() -> cartRepo.find(userId), io);

    return userCf.thenCombine(cartCf, Pair::new)
        .thenCompose(p ->
            pricingService.quoteAsync(p.cart())            // CF<Price>
                .exceptionally(ex -> Price.fallback())      // never fails
                .thenApply(price -> new Checkout(p.user(), p.cart(), price)))
        .thenApply(discountService::applyLoyalty);          // sync tail
}

Read it top-to-bottom: fetch two things in parallel, zip, kick off async pricing with fallback, apply a sync transform. That is the whole grammar.

thenApplyAsync — When to Add the Async Suffix

Every composition operator has an Async variant: thenApplyAsync, thenComposeAsync, etc. The rule:

• No Async suffix — runs on whichever thread completed the previous stage. Cheap for pure transforms.

• Async suffix — resubmits the step to the pool. Use when the step is blocking (DB, HTTP, disk) or CPU-heavy.

cf.thenApply(User::name)                    // fine — cheap
  .thenApplyAsync(this::enrichFromDb, io)   // blocks — must go async on I/O pool
  .thenApply(String::toUpperCase);          // fine — cheap

Passing your own Executor is the difference between a program that scales and one that deadlocks on the common pool. Make it a habit.

allOf and anyOf — Fan-out / Fan-in

Two collection-level operators for the common “wait for N” and “wait for any” cases.

List<CompletableFuture<Product>> lookups = skus.stream()
        .map(sku -> CompletableFuture.supplyAsync(() -> catalog.find(sku), io))
        .toList();

CompletableFuture<List<Product>> all =
    CompletableFuture.allOf(lookups.toArray(CompletableFuture[]::new))
        .thenApply(v -> lookups.stream().map(CompletableFuture::join).toList());

allOf returns CF<Void> — the idiom above is how you recover the results. anyOf returns CF<Object> (untyped, sadly) — useful for racing redundant data sources.

Enter Java 21: Virtual Threads Change the Calculus

Before Java 21, CF chains existed because platform threads are expensive (~1 MB stack each). You could not afford one platform thread per in-flight request. So you wrote non-blocking code with CF or Reactor.

Virtual threads (Project Loom, stable in Java 21) invert this. A virtual thread costs ~500 bytes and blocks cheaply — the JVM unmounts it from the carrier when it blocks on I/O. Suddenly, this synchronous code is fine at scale:

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    Future<User> userF = executor.submit(() -> userRepo.find(id));
    Future<Cart> cartF = executor.submit(() -> cartRepo.find(id));
    return new Checkout(userF.get(), cartF.get());   // blocking is OK now
}

That reads exactly like sync code and scales to millions of concurrent requests. You get to drop most CF chains.

When to Prefer Virtual Threads Over CompletableFuture

Situation	Prefer
Server handling many blocking I/O requests (JDBC, HTTP client, file)	Virtual threads
Composing async APIs that already return CompletableFuture (Spring WebClient, AWS SDK v2)	CompletableFuture
CPU-bound parallel work	Neither — use ForkJoinPool / parallel streams
Fan-out to many services and gather	Virtual threads with structured concurrency (StructuredTaskScope, preview)
Existing reactive codebase (Reactor, RxJava)	Stay reactive; don’t mix paradigms

Structured concurrency (Java 21 preview / 23 second preview) is the real endpoint — it makes virtual-thread fan-out safe and cancellable, and will likely be the enterprise default by the time you finish this roadmap. Phase 05 goes deep.

⚠️ What Most People Get Wrong

They pin a virtual thread with synchronized. If a virtual thread hits a synchronized block that then does blocking I/O, the JVM cannot unmount it — it stays pinned to its carrier and you lose the scaling win. Use ReentrantLock for any critical section that might block. JEP 491 (Java 24) removes most pinning, but until your prod JDK is on 24 the rule stands: no synchronized around I/O in virtual-thread code.

Anti-Patterns to Recognise in Reviews

• .get() in the middle of a chain — blocks a pooled thread. Fix with thenCompose.

• No custom executor — every supplyAsync on the common pool. Under load, everything queues behind everything.

• Swallowed exceptions — a CF that fails silently because nothing called handle/exceptionally/join. Chain .whenComplete((v, ex) -> log(...)) at least at the tail.

• thenApply doing blocking work — the transform runs on the completing thread; blocking here poisons whichever pool completed the previous stage.

• Reinventing allOf — collecting CFs into a list, then looping .join() on each sequentially. That’s serial, not parallel.

Testing Async Code

Two rules keep tests deterministic:

1. Use .join() (not .get()) at the assertion boundary. join() throws unchecked, matches assertion style.

2. Pass a synchronous executor when possible. Runnable::run as the executor makes CF chains fully deterministic in tests without changing production code.

Executor sync = Runnable::run;
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "hi", sync);
assertEquals("hi", cf.join());

For flakier integration tests, Awaitility (await().atMost(2, SECONDS).until(...)) beats Thread.sleep.

Bridge to Phase 05

Phase 05 (Concurrency & Performance) picks up where this file ends:

• Java Memory Model, volatile, happens-before

• ExecutorService design, pool sizing (Little’s Law)

• Virtual threads deep dive, structured concurrency, ScopedValue

• Locks vs AtomicReference vs concurrent collections

• Diagnosing deadlocks, contention, and thread starvation with JFR

For now: understand CF as the functional composition of async values, know the three composition operators, know when virtual threads let you skip CF entirely, and never block a pool thread inside a chain.

What to Practice This Week

1. Take one synchronous service method that does 2-3 sequential I/O calls and rewrite it as a CF chain with thenCombine where possible. Measure latency.

2. Add .exceptionally or handle so every chain has a well-defined failure value — no unhandled CompletionExceptions.

3. Rewrite the same method a third time using virtual threads (Executors.newVirtualThreadPerTaskExecutor()). Compare readability and throughput.

4. Read Brian Goetz’s talk “Java’s Virtual Threads: A Comprehensive Introduction” and JEP 444 (Virtual Threads final).

---

Return to README.md · Previous: 04_optional_done_right.md · Next: projects.md