jdfive

Pipelines with CompletableFuture

Priya can fan out price checks and wait for results. But a full order is a chain of steps: validate the cart, fetch prices, charge the card, send a confirmation email. With Future, each step means another blocking get(). The main thread sits idle, waiting. Priya wants a pipeline — like an assembly line where each station hands work to the next without everyone standing still.

Why CompletableFuture?

Plain Future has limits:

  • No easy way to chain “do A, then B, then C.”
  • Exception handling is clunky.
  • Combining many futures takes boilerplate.
  • Everything blocks on get().

CompletableFuture adds a fluent, chainable API. Think of it as pipes: output from one step flows into the next.

Creating CompletableFuture

Static factory methods

// Already completed with a value
CompletableFuture<String> future = CompletableFuture.completedFuture("Result");

// Already completed with an exception
CompletableFuture<String> failed = CompletableFuture.failedFuture(
    new RuntimeException("Supplier down")
);

// Run async work that returns a value (uses common ForkJoin pool by default)
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Result";
});

// Run async work with your own executor (better for I/O)
ExecutorService executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return fetchPrice(productId);
}, executor);

// Run async work with no return value
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    System.out.println("Task executing");
});

Manual completion

Sometimes another thread finishes the work later:

CompletableFuture<String> future = new CompletableFuture<>();

new Thread(() -> {
    String result = fetchPriceFromSupplier("SKU-42");
    future.complete(result);
}).start();

String result = future.get(); // Blocks until complete() is called

thenApply — transform the result

thenApply takes the result and transforms it synchronously — like a map step:

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "Hello")
    .thenApply(s -> s + " World")
    .thenApply(String::toUpperCase);

String result = future.join(); // "HELLO WORLD"

Use join() like get() but without checked exceptions. At the end of a pipeline it is fine. Prefer callbacks over blocking in the middle when you can.

thenCompose — chain another async step

When the next step is also async (another API call), use thenCompose, not thenApply:

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "Hello")
    .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));

Difference:

  • thenApply — synchronous transformation, returns a plain value.
  • thenCompose — asynchronous transformation, returns another CompletableFuture.

Getting this wrong — using thenApply when the inner function returns a CompletableFuture — gives you a nested mess: CompletableFuture<CompletableFuture<String>>.

thenAccept and thenRun — side effects

thenAccept runs code with the result but does not change the pipeline value:

CompletableFuture
    .supplyAsync(() -> "Order confirmed")
    .thenAccept(msg -> sendEmail(msg));

thenRun runs after completion, with no access to the result:

CompletableFuture
    .supplyAsync(() -> processPayment())
    .thenRun(() -> System.out.println("Payment step finished"));

Async variants

By default, thenApply runs on the same thread that completed the previous step. To run on a thread pool:

ExecutorService executor = Executors.newFixedThreadPool(10);

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "Hello", executor)
    .thenApplyAsync(s -> s + " World", executor)
    .thenApplyAsync(String::toUpperCase, executor);

Use thenApplyAsync / thenComposeAsync when you want each step on a worker thread.

QuickCart order pipeline

Priya wires her order flow as a pipeline:

public CompletableFuture<String> processOrder(String orderId, ExecutorService executor) {
    return CompletableFuture
        .supplyAsync(() -> validateCart(orderId), executor)
        .thenCompose(cart -> CompletableFuture
            .supplyAsync(() -> fetchTotalPrice(cart), executor))
        .thenCompose(total -> CompletableFuture
            .supplyAsync(() -> chargeCard(orderId, total), executor))
        .thenApply(chargeId -> "Order " + orderId + " confirmed. Charge: " + chargeId)
        .exceptionally(ex -> "Order failed: " + ex.getMessage());
}

Each step runs when the previous one finishes. The main thread does not block between steps — it registers what to do next.

Pipeline pattern

CompletableFuture<String> pipeline = CompletableFuture
    .supplyAsync(() -> fetchCart(orderId))
    .thenApply(cart -> validateCart(cart))
    .thenCompose(validated -> saveOrder(validated))
    .thenApply(saved -> buildConfirmation(saved))
    .exceptionally(ex -> handleError(ex));

Combining futures

thenCombine — merge two results

When two independent checks must finish before you proceed:

CompletableFuture<String> priceCheck = CompletableFuture.supplyAsync(() -> "Rs 799");
CompletableFuture<String> stockCheck = CompletableFuture.supplyAsync(() -> "In stock");

CompletableFuture<String> combined = priceCheck.thenCombine(stockCheck,
    (price, stock) -> price + " — " + stock);

String result = combined.join(); // "Rs 799 — In stock"

allOf — wait for all

When Priya checks delivery for ten products at once:

CompletableFuture<Boolean> f1 = CompletableFuture.supplyAsync(() -> checkDelivery("A"));
CompletableFuture<Boolean> f2 = CompletableFuture.supplyAsync(() -> checkDelivery("B"));
CompletableFuture<Boolean> f3 = CompletableFuture.supplyAsync(() -> checkDelivery("C"));

CompletableFuture<Void> allDone = CompletableFuture.allOf(f1, f2, f3);

allDone.thenRun(() -> {
    boolean a = f1.join();
    boolean b = f2.join();
    boolean c = f3.join();
    System.out.println("All checks done: " + a + ", " + b + ", " + c);
});

allOf returns CompletableFuture<Void>. After it completes, call join() on each individual future to read its result.

anyOf — first one wins

CompletableFuture<String> gateway1 = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(3000);
    return "Gateway 1";
});
CompletableFuture<String> gateway2 = CompletableFuture.supplyAsync(() -> {
    Thread.sleep(1000);
    return "Gateway 2"; // Fastest
});

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(gateway1, gateway2);
String result = (String) anyFuture.get(); // "Gateway 2"

Error handling

exceptionally — recover from errors

If any step throws, the pipeline breaks unless you handle it:

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        throw new RuntimeException("Payment gateway down");
    })
    .exceptionally(ex -> {
        return "Recovered: " + ex.getMessage();
    });

String result = future.join(); // "Recovered: Payment gateway down"

exceptionally only runs on failure. It returns a fallback value so the pipeline can continue.

handle — success and failure in one place

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        if (Math.random() > 0.5) {
            throw new RuntimeException("Error");
        }
        return "Success";
    })
    .handle((result, exception) -> {
        if (exception != null) {
            return "Default value";
        }
        return result;
    });

String result = future.get(); // Either "Success" or "Default value"

whenComplete — side effect, does not change the value

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> "Result")
    .whenComplete((result, exception) -> {
        if (exception != null) {
            System.err.println("Error: " + exception.getMessage());
        } else {
            System.out.println("Success: " + result);
        }
    });

QuickCart: parallel delivery checks with error recovery

Priya checks whether each product can ship to a pincode. If one check fails, she uses a safe default:

public class DeliveryChecker {
    private final ExecutorService executor = Executors.newFixedThreadPool(50);

    public Map<String, Boolean> checkAll(Map<String, String> productToPincode) {
        Map<String, CompletableFuture<Boolean>> futures = new HashMap<>();

        for (Map.Entry<String, String> entry : productToPincode.entrySet()) {
            String productId = entry.getKey();
            String pincode = entry.getValue();

            CompletableFuture<Boolean> future = CompletableFuture
                .supplyAsync(() -> checkDelivery(productId, pincode), executor)
                .exceptionally(ex -> {
                    System.err.println("Delivery check failed for " + productId);
                    return false; // Default on error
                });

            futures.put(productId, future);
        }

        // Wait for all with timeout
        CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0]))
            .orTimeout(5, TimeUnit.SECONDS)
            .join();

        Map<String, Boolean> results = new HashMap<>();
        for (Map.Entry<String, CompletableFuture<Boolean>> entry : futures.entrySet()) {
            results.put(entry.getKey(), entry.getValue().join());
        }
        return results;
    }

    private boolean checkDelivery(String productId, String pincode) {
        // Call delivery partner API...
        return true;
    }
}

Timeout and fallback

orTimeout

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> slowSupplierCall())
    .orTimeout(5, TimeUnit.SECONDS)
    .exceptionally(ex -> {
        if (ex instanceof TimeoutException) {
            return "Timeout — using cached price";
        }
        return "Error occurred";
    });

Fallback from a secondary source

CompletableFuture<String> primary = CompletableFuture.supplyAsync(() -> primaryPriceApi());
CompletableFuture<String> fallback = CompletableFuture.supplyAsync(() -> cachedPrice());

CompletableFuture<String> result = primary
    .exceptionally(ex -> fallback.join());

Full checkout example

Stock checks run in parallel. When all finish, the order pipeline starts:

public void checkout(String orderId, List<String> productIds, ExecutorService executor) {
    List<CompletableFuture<Boolean>> stockChecks = productIds.stream()
        .map(id -> CompletableFuture
            .supplyAsync(() -> checkStock(id), executor)
            .exceptionally(ex -> false))
        .toList();

    CompletableFuture.allOf(stockChecks.toArray(new CompletableFuture[0]))
        .thenCompose(v -> processOrder(orderId, executor))
        .thenAccept(msg -> System.out.println(msg));
}

Errors in any stock check default to false instead of crashing everything.

What to remember

  • CompletableFuture chains async steps like a pipeline.
  • supplyAsync starts work. thenApply transforms. thenCompose chains another async call.
  • thenCombine merges two futures. allOf waits for all. anyOf takes the first result.
  • exceptionally gives a fallback when something fails. handle handles both success and failure.
  • whenComplete logs or records side effects without changing the result.
  • Pass a custom executor for I/O-heavy QuickCart work — do not overload the common pool.
  • Use orTimeout for long-running supplier calls.
  • Prefer chaining over blocking on get() in the middle of your code.

What Priya does next: parallel orders work great until two workers try to update the same sold-count at the same time — and the numbers stop adding up.