// The problem, in real BiblioTech code.
ExecutorService executor = Executors.newFixedThreadPool(4);

Future<Catalog> f1 = executor.submit(() -> loadCatalog());
Catalog c = f1.get();                     // BLOCKS. The thread stops here.

Future<Double> f2 = executor.submit(() -> calculateFines(c));
double total = f2.get();                  // BLOCKS again.

Future<Path> f3 = executor.submit(() -> export(total));
Path report = f3.get();                   // And again.

// The same with CompletableFuture: the chain is DECLARED and it returns
// immediately. No thread blocks at any point.
CompletableFuture<Path> report =
        CompletableFuture.supplyAsync(() -> loadCatalog(), ioPool)
                         .thenApplyAsync(c -> calculateFines(c), computePool)
                         .thenApplyAsync(total -> export(total), ioPool)
                         .exceptionally(error -> errorPath(error));

System.out.println("[main] chain launched; the menu is still alive");
showMenu();         // the main thread has NOT waited for anything

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;

// 1. supplyAsync: runs a Supplier and RETURNS a value.
//    It is the most frequent entry point.
CompletableFuture<Catalog> f1 =
        CompletableFuture.supplyAsync(() -> loadCatalog());

// With your own executor (recommended, section 4):
CompletableFuture<Catalog> f2 =
        CompletableFuture.supplyAsync(() -> loadCatalog(), ioPool);

// 2. runAsync: runs a Runnable, returns NO value.
//    The type is CompletableFuture<Void>.
CompletableFuture<Void> f3 =
        CompletableFuture.runAsync(() -> recordAudit("start-up"), ioPool);

// 3. completedFuture: already completed with a value. It runs nothing.
//    Useful for cached values and for tests.
CompletableFuture<Catalog> f4 =
        CompletableFuture.completedFuture(cachedCatalog);

// 4. failedFuture (Java 9): already completed with an exception.
CompletableFuture<Catalog> f5 =
        CompletableFuture.failedFuture(new BiblioTechException("catalogue unavailable"));

// 5. Empty, to complete manually later (section 13).
CompletableFuture<Catalog> f6 = new CompletableFuture<>();

/**
 * ALWAYS returns a CompletableFuture, whether the data is in the
 * cache (immediate answer) or has to be computed (asynchronous).
 * The caller does not have to tell the two cases apart.
 */
public CompletableFuture<Card> getCard(String isbn) {
    Card cached = cache.get(isbn);
    if (cached != null) {
        return CompletableFuture.completedFuture(cached);    // already ready
    }
    return CompletableFuture.supplyAsync(() -> buildCard(isbn), ioPool);
}

// DANGEROUS: I/O on the common pool.
// The common pool has (cores - 1) threads and the whole JVM SHARES it:
// parallel streams (10-04), other libraries and the rest of your
// code. With 7 threads and 10 slow file reads, the pool becomes
// useless for everybody.
CompletableFuture.supplyAsync(() -> Files.readAllLines(hugePath));

// CORRECT: your own executor for I/O, sized according to 08-05.
private static final ExecutorService IO_POOL = new ThreadPoolExecutor(
        16, 16, 0L, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(200),
        r -> new Thread(r, "bibliotech-async-io-" + counter.getAndIncrement()),
        new ThreadPoolExecutor.CallerRunsPolicy());

CompletableFuture.supplyAsync(() -> Files.readAllLines(hugePath), IO_POOL);

import java.util.concurrent.CompletableFuture;

CompletableFuture<Void> chain =
    CompletableFuture
        // 1. Produces a Catalog
        .supplyAsync(() -> loadCatalog(), IO_POOL)

        // 2. thenApply: Catalog -> Double. Transforms.
        .thenApply(catalog -> calculateTotalFines(catalog))

        // 3. thenApply: Double -> String. Another transformation.
        .thenApply(total -> String.format("Fines outstanding: %.2f EUR", total))

        // 4. thenAccept: consumes the String, produces nothing.
        .thenAccept(text -> System.out.println(text))

        // 5. thenRun: receives and returns nothing. For final effects.
        .thenRun(() -> LOG.info("Fines report completed"));

System.out.println("[main] chain declared, carrying on working");

// A method returning an ordinary value:
Double calculateFines(Catalog c) { ... }

// A method returning a CompletableFuture (because it is asynchronous):
CompletableFuture<Double> calculateFinesAsync(Catalog c) { ... }

// THE MISTAKE: thenApply with a function returning a CompletableFuture.
CompletableFuture<CompletableFuture<Double>> nested =
        CompletableFuture.supplyAsync(() -> loadCatalog())
                         .thenApply(c -> calculateFinesAsync(c));

// Now, to reach the Double, you have to unwrap TWICE:
Double d = nested.get().get();       // horrible, and it blocks twice

// THE SOLUTION: thenCompose FLATTENS the nesting.
CompletableFuture<Double> flat =
        CompletableFuture.supplyAsync(() -> loadCatalog())
                         .thenCompose(c -> calculateFinesAsync(c));

Double d2 = flat.join();             // a single level

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.CompletableFuture;

public class AsyncLookup {

    /** Synchronous: returns the value directly. */
    private Material findMaterial(String isbn) { ... }

    /** Asynchronous: returns a CompletableFuture. */
    private CompletableFuture<Card> buildCardAsync(Material m) { ... }

    /** Synchronous: a cheap transformation. */
    private String format(Card c) { ... }

    /**
     * A chain alternating synchronous and asynchronous operations.
     * Notice that thenCompose appears exactly where the
     * function returns a CompletableFuture.
     */
    public CompletableFuture<String> formattedCard(String isbn) {
        return CompletableFuture
                .supplyAsync(() -> findMaterial(isbn), IO_POOL)    // -> Material
                .thenCompose(this::buildCardAsync)                 // -> Card (async)
                .thenApply(this::format);                          // -> String (sync)
    }
}

.thenApply(f)                  // 1. no suffix
.thenApplyAsync(f)             // 2. with suffix, default executor
.thenApplyAsync(f, executor)   // 3. with suffix and explicit executor

public class WhichThreadRuns {

    public static void main(String[] args) throws Exception {

        ExecutorService pool = Executors.newFixedThreadPool(2,
                r -> new Thread(r, "bibliotech-pool"));

        System.out.println("main runs on: " + Thread.currentThread().getName());

        CompletableFuture<String> f = CompletableFuture
                .supplyAsync(() -> {
                    trace("supplyAsync");
                    return "catalog";
                }, pool)
                .thenApply(v -> {
                    trace("thenApply");          // the pool's thread
                    return v + "-processed";
                })
                .thenApplyAsync(v -> {
                    trace("thenApplyAsync");     // commonPool
                    return v + "-async";
                })
                .thenApplyAsync(v -> {
                    trace("thenApplyAsync(pool)"); // the pool we give it
                    return v + "-own";
                }, pool);

        System.out.println("result: " + f.join());
        pool.shutdown();
    }

    static void trace(String stage) {
        System.out.printf("  %-24s -> %s%n", stage, Thread.currentThread().getName());
    }
}

main runs on: main
  supplyAsync              -> bibliotech-pool
  thenApply                -> bibliotech-pool
  thenApplyAsync           -> ForkJoinPool.commonPool-worker-1
  thenApplyAsync(pool)     -> bibliotech-pool
result: catalog-processed-async-own

// WRONG: blocking in a suffix-free stage occupies the previous thread,
// which could be a common-pool thread or main itself.
.thenApply(catalog -> writeToDisk(catalog))           // 500 ms blocking

// RIGHT: the heavy work goes to a dedicated pool.
.thenApplyAsync(catalog -> writeToDisk(catalog), IO_POOL)

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.CompletableFuture;

public class AsyncSummary {

    /**
     * The two queries are independent and are launched at the same time.
     * The total time is that of the SLOWEST, not the sum.
     */
    public CompletableFuture<String> fullSummary(String isbn) {

        CompletableFuture<Material> material =
                CompletableFuture.supplyAsync(() -> catalog.find(isbn), IO_POOL);

        CompletableFuture<Integer> loans =
                CompletableFuture.supplyAsync(() -> registry.countLoans(isbn), IO_POOL);

        // thenCombine waits for BOTH and applies the BiFunction (04-06).
        return material.thenCombine(loans, (m, n) ->
                String.format("%s (%s) - %d historical loans",
                        m.title(), m.isbn(), n));
    }

    /** Three or more: the thenCombines are chained. */
    public CompletableFuture<FullReport> fullReport(String isbn) {

        CompletableFuture<Material> material =
                CompletableFuture.supplyAsync(() -> catalog.find(isbn), IO_POOL);
        CompletableFuture<Integer> loans =
                CompletableFuture.supplyAsync(() -> registry.countLoans(isbn), IO_POOL);
        CompletableFuture<Double> fines =
                CompletableFuture.supplyAsync(() -> calculator.finesOf(isbn), IO_POOL);

        return material
                .thenCombine(loans, MaterialLoansPartial::new)
                .thenCombine(fines, (partial, f) ->
                        new FullReport(partial.material(), partial.loans(), f));
    }

    private record MaterialLoansPartial(Material material, int loans) { }
}

package com.nexussoftware.bibliotech.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;

public class AsyncNotices {

    /**
     * Sends every notice in parallel and waits for them ALL to finish.
     *
     * The collection pattern is always the same:
     *   1. Create the list of futures.
     *   2. allOf(...) over the array.
     *   3. thenApply walking the futures and calling join()
     *      —safe, because allOf guarantees they are already complete—.
     */
    public CompletableFuture<List<NoticeResult>> sendAll(List<Loan> overdue) {

        List<CompletableFuture<NoticeResult>> futures = new ArrayList<>();
        for (Loan l : overdue) {
            futures.add(CompletableFuture.supplyAsync(() -> sendNotice(l), IO_POOL));
        }

        // allOf takes an array, not a list.
        CompletableFuture<Void> all =
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

        return all.thenApply(v -> {
            List<NoticeResult> results = new ArrayList<>(futures.size());
            for (CompletableFuture<NoticeResult> f : futures) {
                // join() here does NOT block: allOf has already guaranteed
                // that they are all complete.
                results.add(f.join());
            }
            return results;
        });
    }

    /**
     * FAULT-TOLERANT variant: one failed notice does not take down the set.
     *
     * CAREFUL: if one future fails, the allOf fails too and its join()
     * would rethrow the exception. The solution is to shield EACH future with
     * its own exceptionally BEFORE aggregating them.
     */
    public CompletableFuture<List<NoticeResult>> sendAllTolerant(
            List<Loan> overdue) {

        List<CompletableFuture<NoticeResult>> futures = new ArrayList<>();
        for (Loan l : overdue) {
            futures.add(CompletableFuture
                    .supplyAsync(() -> sendNotice(l), IO_POOL)
                    .exceptionally(e -> NoticeResult.failure(l, e.getMessage())));
        }

        return CompletableFuture
                .allOf(futures.toArray(new CompletableFuture[0]))
                .thenApply(v -> {
                    List<NoticeResult> r = new ArrayList<>();
                    for (CompletableFuture<NoticeResult> f : futures) r.add(f.join());
                    return r;
                });
    }
}

// Three sources for the same catalogue; the first will do.
CompletableFuture<Catalog> cache  = readFromCache();
CompletableFuture<Catalog> file   = readFromMainFile();
CompletableFuture<Catalog> backup = readFromBackup();

CompletableFuture<Object> first =
        CompletableFuture.anyOf(cache, file, backup);

Catalog c = (Catalog) first.join();          // a cast is needed

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.CompletableFuture;
import java.util.logging.Level;
import java.util.logging.Logger;

public class AsyncErrorHandling {

    private static final Logger LOG =
            Logger.getLogger(AsyncErrorHandling.class.getName());

    /** 1. exceptionally: a fallback value if something fails. */
    public CompletableFuture<Catalog> loadWithFallback() {
        return CompletableFuture
                .supplyAsync(() -> loadFromFile(), IO_POOL)
                .exceptionally(error -> {
                    // 'error' is a CompletionException WRAPPING the cause.
                    LOG.log(Level.WARNING, "Load failed; using the fallback",
                            error.getCause());
                    return Catalog.empty();       // the chain CONTINUES
                });
    }

    /** 2. handle: it always runs, and unifies the two paths. */
    public CompletableFuture<String> reportWithHandle() {
        return CompletableFuture
                .supplyAsync(() -> generateReport(), IO_POOL)
                .handle((result, error) -> {
                    // Exactly one of the two is null.
                    if (error != null) {
                        LOG.log(Level.SEVERE, "Report failed", error);
                        return "REPORT UNAVAILABLE: " + causeOf(error).getMessage();
                    }
                    return "REPORT OK: " + result;
                });
    }

    /** 3. whenComplete: observes without altering. Ideal for traces and metrics. */
    public CompletableFuture<Catalog> loadWithTrace() {
        long start = System.nanoTime();
        return CompletableFuture
                .supplyAsync(() -> loadFromFile(), IO_POOL)
                .whenComplete((catalog, error) -> {
                    long ms = (System.nanoTime() - start) / 1_000_000;
                    if (error != null) {
                        LOG.log(Level.WARNING, "Load failed in " + ms + " ms", error);
                    } else {
                        LOG.log(Level.INFO, "Catalogue loaded in {0} ms ({1} materials)",
                                new Object[] { ms, catalog.size() });
                    }
                    // Returning something here would NOT change the result:
                    // whenComplete only OBSERVES. The error keeps propagating.
                });
    }

    /** Utility: unwrap the real cause. */
    static Throwable causeOf(Throwable t) {
        return (t instanceof java.util.concurrent.CompletionException
                || t instanceof java.util.concurrent.ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }
}

CompletableFuture<Path> report = CompletableFuture
        .supplyAsync(() -> loadCatalog(), IO_POOL)
        .thenApplyAsync(this::calculateFines, COMPUTE_POOL)
        .thenApplyAsync(this::export, IO_POOL)
        .whenComplete((path, error) -> recordMetric(path, error))       // observe
        .exceptionally(error -> {                                       // handle
            LOG.log(Level.SEVERE, "Report chain failed", error);
            return Path.of("reports/error.txt");
        });

public class ErrorPropagation {

    public static void main(String[] args) {

        CompletableFuture<String> chain = CompletableFuture
                .supplyAsync(() -> {
                    System.out.println("  stage 1: OK");
                    return "catalog";
                })
                .thenApply(v -> {
                    System.out.println("  stage 2: throwing an exception");
                    throw new IllegalStateException("corrupt catalogue");
                })
                .thenApply(v -> {
                    System.out.println("  stage 3: DOES NOT RUN");
                    return v + "-processed";
                })
                .thenApply(v -> {
                    System.out.println("  stage 4: NOR DOES THIS");
                    return v.toUpperCase();
                })
                .exceptionally(error -> {
                    System.out.println("  handler: caught " + error.getClass().getSimpleName());
                    System.out.println("           real cause: "
                            + error.getCause().getClass().getSimpleName()
                            + ": " + error.getCause().getMessage());
                    return "FALLBACK-VALUE";
                })
                .thenApply(v -> {
                    System.out.println("  stage 5: this DOES run, with the fallback");
                    return v.toLowerCase();
                });

        System.out.println("result: " + chain.join());
    }
}

  stage 1: OK
  stage 2: throwing an exception
  handler: caught CompletionException
           real cause: IllegalStateException: corrupt catalogue
  stage 5: this DOES run, with the fallback
result: fallback-value

// TRAP: it looks as if the exceptionally protects the whole chain.
CompletableFuture<Path> f = CompletableFuture
        .supplyAsync(() -> loadCatalog())
        .exceptionally(e -> Catalog.empty())        // protects what is ABOVE
        .thenApply(c -> calculateFines(c))          // <-- UNprotected
        .thenApply(m -> export(m));                 // <-- UNprotected

// If calculateFines or export fail, the exception comes out of join()
// and nobody handles it. The exceptionally was too far up.

// CORRECT: the handler at the END of the chain.
CompletableFuture<Path> g = CompletableFuture
        .supplyAsync(() -> loadCatalog())
        .thenApply(c -> calculateFines(c))
        .thenApply(m -> export(m))
        .exceptionally(e -> {                       // protects EVERYTHING above
            LOG.log(Level.SEVERE, "Chain failed", e);
            return Path.of("reports/error.txt");
        });

// If this chain fails, NOTHING VISIBLE HAPPENS. There is no trace,
// no log, no exception. The failure simply disappears.
CompletableFuture.supplyAsync(() -> loadCatalog())
                 .thenAccept(c -> globalCatalog = c);

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

// orTimeout: if it does not complete in time, it FAILS with TimeoutException.
CompletableFuture<Catalog> withDeadline = CompletableFuture
        .supplyAsync(() -> loadCatalog(), IO_POOL)
        .orTimeout(5, TimeUnit.SECONDS)
        .exceptionally(error -> {
            if (causeOf(error) instanceof java.util.concurrent.TimeoutException) {
                LOG.warning("The load exceeded 5 seconds");
            }
            return Catalog.empty();
        });

// completeOnTimeout: if it does not complete in time, it completes
// with the DEFAULT VALUE. It does not fail.
CompletableFuture<Catalog> withDefault = CompletableFuture
        .supplyAsync(() -> loadCatalog(), IO_POOL)
        .completeOnTimeout(Catalog.empty(), 5, TimeUnit.SECONDS);

package com.nexussoftware.bibliotech.persistence;

import java.util.concurrent.CompletableFuture;

public class CallbackAdapter {

    /**
     * An old callback-based API. It is awkward to compose:
     * nesting three of these produces the "pyramid of doom".
     */
    interface CallbackReader {
        void readAsync(String isbn, Callback<Material> callback);
    }

    interface Callback<T> {
        void onComplete(T result);
        void onFailure(Throwable error);
    }

    private final CallbackReader reader;

    public CallbackAdapter(CallbackReader reader) { this.reader = reader; }

    /**
     * Turns the callback API into a composable CompletableFuture.
     * It is the standard pattern for modernising legacy code without touching it.
     */
    public CompletableFuture<Material> read(String isbn) {

        // 1. An empty future, with no task attached.
        CompletableFuture<Material> future = new CompletableFuture<>();

        // 2. The operation is launched with a callback that completes it.
        reader.readAsync(isbn, new Callback<Material>() {
            @Override
            public void onComplete(Material m) {
                future.complete(m);                    // success
            }
            @Override
            public void onFailure(Throwable error) {
                future.completeExceptionally(error);   // failure
            }
        });

        // 3. It returns immediately, without waiting for the callback.
        return future;
    }

    /**
     * And now the old API is composable like any other:
     * three reads in parallel and combined, with no nesting.
     */
    public CompletableFuture<String> compareThree(String i1, String i2, String i3) {
        return read(i1)
                .thenCombine(read(i2), (a, b) -> a.title() + " / " + b.title())
                .thenCombine(read(i3), (pair, c) -> pair + " / " + c.title());
    }
}

CompletableFuture<Catalog> f =
        CompletableFuture.supplyAsync(() -> loadCatalogSlowly(), IO_POOL);

TimeUnit.SECONDS.sleep(1);
boolean cancelled = f.cancel(true);      // the 'true' is IGNORED

System.out.println("cancel(): " + cancelled);             // true
System.out.println("isCancelled(): " + f.isCancelled());  // true
// BUT: the IO_POOL thread carries on running loadCatalogSlowly()
// to the end. The mayInterruptIfRunning parameter DOES NOTHING.

package com.nexussoftware.bibliotech.persistence;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * REAL cancellation of an asynchronous task: a shared flag
 * the task checks. It is the cooperative interruption of 08-02
 * adapted to CompletableFuture, which cannot interrupt on its own.
 */
public class CancellableImport {

    private final AtomicBoolean cancelled = new AtomicBoolean(false);
    private final CompletableFuture<Report> future;

    public CancellableImport(Path file) {
        this.future = CompletableFuture.supplyAsync(() -> importFile(file), IO_POOL);
    }

    private Report importFile(Path file) {
        int processed = 0;
        for (String line : readLines(file)) {

            // An explicit CANCELLATION POINT.
            if (cancelled.get() || Thread.currentThread().isInterrupted()) {
                throw new CancellationException(
                        "Import cancelled after " + processed + " lines");
            }
            process(line);
            processed++;
        }
        return new Report(processed);
    }

    /** Cancellation that DOES stop the work. */
    public void cancel() {
        cancelled.set(true);          // the task will see it and abort
        future.cancel(false);         // and the future completes right away
    }

    public CompletableFuture<Report> future() { return future; }
}

// TERRIBLE: join() inside a stage blocks a pool thread.
// With enough chains like this, the pool runs dry and everything stops.
.thenApply(catalog -> {
    Double fines = calculateFinesAsync(catalog).join();     // BLOCKS
    return fines;
})

// CORRECT: thenCompose flattens without blocking anybody.
.thenCompose(catalog -> calculateFinesAsync(catalog))

// get(): throws ExecutionException and InterruptedException, both CHECKED.
try {
    Catalog c = future.get();
} catch (ExecutionException | InterruptedException e) { ... }

// join(): throws CompletionException, UNchecked. More convenient
// inside lambdas, where a checked exception does not compile.
Catalog c = future.join();

// WRONG: a single twenty-line expression.

// RIGHT: every step with a name.
public CompletableFuture<Path> generateMonthlyReport() {
    return loadCatalog()
            .thenCompose(this::enrichWithLoans)
            .thenApplyAsync(this::calculateFines, COMPUTE_POOL)
            .thenApplyAsync(this::format, COMPUTE_POOL)
            .thenApplyAsync(this::writeFile, IO_POOL)
            .orTimeout(60, TimeUnit.SECONDS)
            .whenComplete(this::recordMetrics)
            .exceptionally(this::errorReport);
}

package com.nexussoftware.bibliotech.service;

import com.nexussoftware.bibliotech.domain.*;

import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Asynchronous generation of BiblioTech's monthly report.
 *
 * The complete chain:
 *   1. Load the catalogue (I/O)      -- in parallel with 2
 *   2. Load the active loans (I/O)   -- in parallel with 1
 *   3. Combine both
 *   4. Calculate the fines (CPU)
 *   5. Format the report (CPU)
 *   6. Write the file (I/O)
 *
 * NO thread blocks waiting: the menu carries on serving the user
 * throughout the whole operation.
 */
public class AsyncReportGenerator implements AutoCloseable {

    private static final Logger LOG =
            Logger.getLogger(AsyncReportGenerator.class.getName());

    /** I/O pool: many threads because it is nearly all waiting (08-01). */
    private final ExecutorService ioPool;
    /** Compute pool: one thread per core. */
    private final ExecutorService computePool;

    private final ConcurrentCatalog catalog;
    private final SafeLoanRegistry registry;
    private final FineCalculator calculator;

    public AsyncReportGenerator(ConcurrentCatalog catalog,
                                SafeLoanRegistry registry,
                                FineCalculator calculator) {
        this.catalog = catalog;
        this.registry = registry;
        this.calculator = calculator;

        AtomicInteger nIo = new AtomicInteger(1);
        this.ioPool = new ThreadPoolExecutor(
                16, 16, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(200),
                r -> new Thread(r, "bibliotech-async-io-" + nIo.getAndIncrement()),
                new ThreadPoolExecutor.CallerRunsPolicy());

        AtomicInteger nCpu = new AtomicInteger(1);
        this.computePool = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors(),
                r -> new Thread(r, "bibliotech-async-cpu-" + nCpu.getAndIncrement()));
    }

    // ---------- STAGES ----------

    private CompletableFuture<List<Material>> loadMaterials() {
        return CompletableFuture.supplyAsync(() -> {
            trace("loading materials");
            pause(400);                        // simulated I/O
            return catalog.byType(MaterialType.BOOK);
        }, ioPool);
    }

    private CompletableFuture<List<Loan>> loadActiveLoans() {
        return CompletableFuture.supplyAsync(() -> {
            trace("loading loans");
            pause(500);                        // simulated I/O, IN PARALLEL
            return registry.allActive();
        }, ioPool);
    }

    /** A CPU stage: it is sent to the compute pool explicitly. */
    private ReportData compute(List<Material> materials, List<Loan> loans) {
        trace("calculating fines");
        double total = 0;
        int overdue = 0;
        for (Loan l : loans) {
            double f = calculator.calculate(l);
            if (f > 0) { total += f; overdue++; }
        }
        return new ReportData(materials.size(), loans.size(), overdue, total);
    }

    private String format(ReportData d) {
        trace("formatting");
        return """
               ===========================================
                 BiblioTech - Monthly fines report
                 Nexus Software
               ===========================================
                 Materials in catalogue : %d
                 Active loans           : %d
                 Overdue loans          : %d
                 Accumulated fines      : %.2f EUR
               ===========================================
               """.formatted(d.materials(), d.loans(), d.overdue(), d.fines());
    }

    private Path write(String content) {
        trace("writing the file");
        pause(300);                            // simulated I/O
        Path target = Path.of("reports", "monthly-fines.txt");
        // In the real code: AtomicWrite.write(target, content) from 07-06
        return target;
    }

    // ---------- THE CHAIN ----------

    /**
     * Declares the whole chain and RETURNS IMMEDIATELY.
     * The calling thread waits for nothing.
     */
    public CompletableFuture<Path> generate() {

        long start = System.nanoTime();

        // 1 and 2 start AT THE SAME TIME: they are not chained, they are combined.
        CompletableFuture<List<Material>> materials = loadMaterials();
        CompletableFuture<List<Loan>> loans = loadActiveLoans();

        return materials
                // 3. Wait for both and combine them. The work is sent to the
                //    compute pool with thenCombineAsync, so as not to occupy
                //    an I/O thread with CPU work.
                .thenCombineAsync(loans, this::compute, computePool)

                // 4. Format: CPU, same pool.
                .thenApplyAsync(this::format, computePool)

                // 5. Write: I/O, I/O pool.
                .thenApplyAsync(this::write, ioPool)

                // 6. Global deadline.
                .orTimeout(30, TimeUnit.SECONDS)

                // 7. Observe without altering: metrics.
                .whenComplete((path, error) -> {
                    long ms = (System.nanoTime() - start) / 1_000_000;
                    if (error == null) {
                        LOG.log(Level.INFO, "Report generated in {0} ms: {1}",
                                new Object[] { ms, path });
                    } else {
                        LOG.log(Level.SEVERE, "Report failed after " + ms + " ms", error);
                    }
                })

                // 8. Handle: at the END, to cover the whole chain.
                .exceptionally(error -> {
                    Throwable cause = causeOf(error);
                    if (cause instanceof TimeoutException) {
                        LOG.warning("The report exceeded 30 seconds");
                    }
                    return Path.of("reports", "report-unavailable.txt");
                });
    }

    // ---------- HELPERS ----------

    record ReportData(int materials, int loans, int overdue, double fines) { }

    static Throwable causeOf(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    private static void trace(String stage) {
        System.out.printf("      [%s] %s%n", Thread.currentThread().getName(), stage);
    }

    private static void pause(long ms) {
        try { TimeUnit.MILLISECONDS.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    @Override
    public void close() {
        for (ExecutorService ex : List.of(computePool, ioPool)) {
            ex.shutdown();
            try {
                if (!ex.awaitTermination(15, TimeUnit.SECONDS)) ex.shutdownNow();
            } catch (InterruptedException e) {
                ex.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
    }
}

package com.nexussoftware.bibliotech.presentation;

import java.nio.file.Path;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class MenuWithAsyncReport {

    public static void main(String[] args) throws Exception {

        try (AsyncReportGenerator generator = new AsyncReportGenerator(
                catalog, registry, calculator)) {

            System.out.println("=== BiblioTech - Nexus Software ===");
            System.out.println("Launching the monthly report in the background...\n");

            // The whole chain is declared and returns instantly.
            CompletableFuture<Path> report = generator.generate();

            // We register what to do when it finishes. We do NOT wait.
            report.thenAccept(path ->
                    System.out.println("\n>>> Report ready at: " + path));

            // The menu carries on serving the user.
            for (int i = 1; i <= 8; i++) {
                System.out.printf("  [menu] option %d served (the menu is NOT blocked)%n", i);
                TimeUnit.MILLISECONDS.sleep(200);
            }

            // Only at the end, if the result is genuinely needed, do we wait.
            Path path = report.get(30, TimeUnit.SECONDS);
            System.out.println("\n[main] confirmed: " + path);
        }
    }
}

=== BiblioTech - Nexus Software ===
Launching the monthly report in the background...

  [menu] option 1 served (the menu is NOT blocked)
      [bibliotech-async-io-1] loading materials
      [bibliotech-async-io-2] loading loans
  [menu] option 2 served (the menu is NOT blocked)
  [menu] option 3 served (the menu is NOT blocked)
      [bibliotech-async-cpu-1] calculating fines
      [bibliotech-async-cpu-1] formatting
  [menu] option 4 served (the menu is NOT blocked)
      [bibliotech-async-io-3] writing the file

>>> Report ready at: reports/monthly-fines.txt
  [menu] option 5 served (the menu is NOT blocked)
  [menu] option 6 served (the menu is NOT blocked)
  [menu] option 7 served (the menu is NOT blocked)
  [menu] option 8 served (the menu is NOT blocked)
INFO: Report generated in 812 ms: reports/monthly-fines.txt

[main] confirmed: reports/monthly-fines.txt

// With CompletableFuture: asynchronous, composable, and demanding to read.
CompletableFuture<Path> f = CompletableFuture
        .supplyAsync(() -> loadCatalog(), ioPool)
        .thenApplyAsync(this::calculateFines, cpuPool)
        .thenApplyAsync(this::export, ioPool)
        .exceptionally(this::fallback);

// With virtual threads: ordinary SEQUENTIAL code, with ordinary try/catch,
// blocking a virtual thread —whose cost is in nanoseconds—.
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    executor.submit(() -> {
        Catalog c = loadCatalog();          // "blocks", but it is dirt cheap
        double f2 = calculateFines(c);
        return export(f2);
    });
}

Future<Catalog> f1 = executor.submit(() -> loadCatalog());
Catalog c = f1.get();
Future<List<Loan>> f2 = executor.submit(() -> loadLoans(c));
List<Loan> p = f2.get();
Future<Double> f3 = executor.submit(() -> calculateFines(p));
double total = f3.get();
Future<Path> f4 = executor.submit(() -> export(total));
Path report = f4.get();

package com.nexussoftware.bibliotech.service;

import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ReportChain {

    private final ExecutorService ioPool;
    private final ExecutorService cpuPool;

    public ReportChain() {
        AtomicInteger a = new AtomicInteger(1);
        this.ioPool = Executors.newFixedThreadPool(8,
                r -> new Thread(r, "report-io-" + a.getAndIncrement()));
        AtomicInteger b = new AtomicInteger(1);
        this.cpuPool = Executors.newFixedThreadPool(
                Runtime.getRuntime().availableProcessors(),
                r -> new Thread(r, "report-cpu-" + b.getAndIncrement()));
    }

    // --- Simulated operations ---

    private Catalog loadCatalog() {
        pause(300);
        return new Catalog();
    }

    /**
     * ASYNCHRONOUS: it returns a CompletableFuture, not a value.
     * That is why it is chained with thenCompose and not thenApply.
     */
    private CompletableFuture<List<Loan>> loadLoansAsync(Catalog c) {
        return CompletableFuture.supplyAsync(() -> {
            pause(300);
            return List.<Loan>of();
        }, ioPool);
    }

    private double calculateFines(List<Loan> p) {
        pause(300);
        return 137.50;
    }

    private Path export(double total) {
        pause(300);
        return Path.of("reports/fines.txt");
    }

    // --- BLOCKING VERSION (the original) ---

    public Path futureVersion() throws Exception {
        ExecutorService ex = Executors.newFixedThreadPool(4);
        try {
            Future<Catalog> f1 = ex.submit(this::loadCatalog);
            Catalog c = f1.get();                                  // BLOCKS

            Future<List<Loan>> f2 = ex.submit(() -> {
                pause(300); return List.<Loan>of();
            });
            List<Loan> p = f2.get();                               // BLOCKS

            Future<Double> f3 = ex.submit(() -> calculateFines(p));
            double total = f3.get();                               // BLOCKS

            Future<Path> f4 = ex.submit(() -> export(total));
            return f4.get();                                       // BLOCKS
        } finally {
            ex.shutdown();
        }
    }

    // --- ASYNCHRONOUS VERSION ---

    public CompletableFuture<Path> completableFutureVersion() {

        long start = System.nanoTime();

        return CompletableFuture
                // 1. I/O: the I/O pool
                .supplyAsync(this::loadCatalog, ioPool)

                // 2. thenCompose because loadLoansAsync returns
                //    a CompletableFuture. With thenApply we would get
                //    CompletableFuture<CompletableFuture<List<Loan>>>.
                .thenCompose(this::loadLoansAsync)

                // 3. CPU: the compute pool
                .thenApplyAsync(this::calculateFines, cpuPool)

                // 4. I/O: the I/O pool
                .thenApplyAsync(this::export, ioPool)

                // 5. Global deadline
                .orTimeout(20, TimeUnit.SECONDS)

                // 6. Observe without altering
                .whenComplete((path, error) -> {
                    long ms = (System.nanoTime() - start) / 1_000_000;
                    System.out.printf("  [chain] finished in %d ms (%s)%n",
                            ms, error == null ? "OK" : "ERROR");
                })

                // 7. Handle at the END: it covers every earlier stage
                .exceptionally(error -> {
                    System.err.println("  [chain] failure: " + causeOf(error).getMessage());
                    return Path.of("reports/unavailable.txt");
                });
    }

    static Throwable causeOf(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    static void pause(long ms) {
        try { TimeUnit.MILLISECONDS.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    public void shutdown() {
        ioPool.shutdown();
        cpuPool.shutdown();
    }

    public static void main(String[] args) throws Exception {
        ReportChain rc = new ReportChain();
        try {
            System.out.println("=== FUTURE VERSION (blocking) ===");
            long t1 = System.nanoTime();
            Path p1 = rc.futureVersion();
            long ms1 = (System.nanoTime() - t1) / 1_000_000;
            System.out.println("  result: " + p1 + "  (" + ms1 + " ms)");
            System.out.println("  the main thread was blocked 4 times\n");

            System.out.println("=== COMPLETABLEFUTURE VERSION ===");
            long t2 = System.nanoTime();
            CompletableFuture<Path> f = rc.completableFutureVersion();
            long msDeclaration = (System.nanoTime() - t2) / 1_000_000;
            System.out.println("  chain declared in " + msDeclaration + " ms");
            System.out.println("  the main thread is still free; doing other things...");

            for (int i = 1; i <= 5; i++) {
                System.out.println("    main work " + i);
                TimeUnit.MILLISECONDS.sleep(150);
            }

            Path p2 = f.get(20, TimeUnit.SECONDS);
            long ms2 = (System.nanoTime() - t2) / 1_000_000;
            System.out.println("  result: " + p2 + "  (" + ms2 + " ms)");
        } finally {
            rc.shutdown();
        }
    }
}

=== FUTURE VERSION (blocking) ===
  result: reports/fines.txt  (1214 ms)
  the main thread was blocked 4 times

=== COMPLETABLEFUTURE VERSION ===
  chain declared in 3 ms
  the main thread is still free; doing other things...
    main work 1
    main work 2
    main work 3
    main work 4
    main work 5
  [chain] finished in 1208 ms (OK)
  result: reports/fines.txt  (1211 ms)

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class CardService implements AutoCloseable {

    record MaterialData(String title, String author, String isbn) {
        static MaterialData unknown(String isbn) {
            return new MaterialData("(title not available)", "(author unknown)", isbn);
        }
    }

    record FullCard(MaterialData material, int loans, double rating) {
        static FullCard minimal(String isbn) {
            return new FullCard(MaterialData.unknown(isbn), -1, -1.0);
        }
        @Override public String toString() {
            return String.format("%s / %s [%s] - %s loans - rating %s",
                    material.title(), material.author(), material.isbn(),
                    loans < 0 ? "?" : loans,
                    rating < 0 ? "?" : String.format("%.1f", rating));
        }
    }

    private final ExecutorService pool;
    private final boolean failRatings;

    public CardService(boolean failRatings) {
        this.failRatings = failRatings;
        AtomicInteger n = new AtomicInteger(1);
        this.pool = Executors.newFixedThreadPool(8,
                r -> new Thread(r, "cards-" + n.getAndIncrement()));
    }

    // --- The three queries, each with its own fallback ---

    private CompletableFuture<MaterialData> queryMaterial(String isbn) {
        return CompletableFuture.supplyAsync(() -> {
            pause(400);
            return new MaterialData("Effective Java", "J. Bloch", isbn);
        }, pool)
        // An INDIVIDUAL fallback: if this query fails, the set
        // carries on with a degraded value.
        .exceptionally(e -> {
            System.err.println("  [fallback] material not available: " + causeOf(e).getMessage());
            return MaterialData.unknown(isbn);
        });
    }

    private CompletableFuture<Integer> queryLoans(String isbn) {
        return CompletableFuture.supplyAsync(() -> {
            pause(600);                        // the slowest: it sets the total
            return 47;
        }, pool)
        .exceptionally(e -> -1);
    }

    private CompletableFuture<Double> queryRating(String isbn) {
        return CompletableFuture.supplyAsync(() -> {
            pause(300);
            if (failRatings) {
                throw new IllegalStateException("ratings service down");
            }
            return 4.6;
        }, pool)
        .exceptionally(e -> {
            System.err.println("  [fallback] rating not available: "
                    + causeOf(e).getMessage());
            return -1.0;
        });
    }

    // --- The combination ---

    public CompletableFuture<FullCard> fullCard(String isbn) {

        // All THREE are launched at once: the assignment already starts the work.
        CompletableFuture<MaterialData> material = queryMaterial(isbn);
        CompletableFuture<Integer>      loans    = queryLoans(isbn);
        CompletableFuture<Double>       rating   = queryRating(isbn);

        return material
                // Chained thenCombine: first material+loans...
                .thenCombine(loans, (m, l) -> new Object[] { m, l })
                // ...and then the pair with the rating.
                .thenCombine(rating, (pair, r) -> new FullCard(
                        (MaterialData) pair[0], (Integer) pair[1], r))
                // If the set takes too long, a minimal card instead of a failure.
                .completeOnTimeout(FullCard.minimal(isbn), 2, TimeUnit.SECONDS);
    }

    static Throwable causeOf(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    static void pause(long ms) {
        try { TimeUnit.MILLISECONDS.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    @Override public void close() { pool.shutdown(); }

    public static void main(String[] args) throws Exception {

        System.out.println("=== NORMAL CASE ===");
        try (CardService s = new CardService(false)) {
            long t = System.nanoTime();
            FullCard c = s.fullCard("978-0000000001").get();
            System.out.println("  " + c);
            System.out.printf("  time: %d ms (sequential would be 1300)%n",
                    (System.nanoTime() - t) / 1_000_000);
        }

        System.out.println("\n=== WITH THE RATINGS SERVICE DOWN ===");
        try (CardService s = new CardService(true)) {
            long t = System.nanoTime();
            FullCard c = s.fullCard("978-0000000001").get();
            System.out.println("  " + c);
            System.out.printf("  time: %d ms%n", (System.nanoTime() - t) / 1_000_000);
            System.out.println("  The card is built anyway: degradation, not failure (06-07).");
        }
    }
}

=== NORMAL CASE ===
  Effective Java / J. Bloch [978-0000000001] - 47 loans - rating 4.6
  time: 612 ms (sequential would be 1300)

=== WITH THE RATINGS SERVICE DOWN ===
  [fallback] rating not available: ratings service down
  Effective Java / J. Bloch [978-0000000001] - 47 loans - rating ?
  time: 608 ms
  The card is built anyway: degradation, not failure (06-07).

package com.nexussoftware.bibliotech.service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

public class AsyncNoticeSending implements AutoCloseable {

    record Notice(String id, String employee) { }

    record NoticeResult(String id, boolean successful, String detail) {
        static NoticeResult success(Notice n) {
            return new NoticeResult(n.id(), true, "sent");
        }
        static NoticeResult failure(Notice n, String reason) {
            return new NoticeResult(n.id(), false, reason);
        }
    }

    record SendSummary(int total, int sent, int failed, long ms) {
        double percentage() { return total == 0 ? 0 : 100.0 * sent / total; }
    }

    private final ExecutorService ioPool;
    private final LongAdder completed = new LongAdder();

    public AsyncNoticeSending() {
        AtomicInteger n = new AtomicInteger(1);
        this.ioPool = new ThreadPoolExecutor(
                16, 16, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(300),
                r -> new Thread(r, "bibliotech-notice-" + n.getAndIncrement()),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    /** An individual send: 250 ms of I/O and a simulated failure in some. */
    private NoticeResult send(Notice n) {
        try {
            TimeUnit.MILLISECONDS.sleep(250);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new CompletionException(e);
        }
        if (n.id().hashCode() % 29 == 0) {
            throw new IllegalStateException("employee has no contact address");
        }
        return NoticeResult.success(n);
    }

    /** The complete chain. It returns immediately. */
    public CompletableFuture<SendSummary> sendAll(List<Notice> notices) {

        long start = System.nanoTime();
        final int total = notices.size();

        // 1-2. One future per notice, SHIELDED with its own exceptionally.
        //      Without that shield, a single failure would fail the whole allOf.
        List<CompletableFuture<NoticeResult>> futures = new ArrayList<>(total);
        for (Notice n : notices) {
            futures.add(CompletableFuture
                    .supplyAsync(() -> send(n), ioPool)
                    .exceptionally(e -> NoticeResult.failure(n, causeOf(e).getMessage()))
                    // Counted on finishing, in success or failure:
                    // that way main can show progress without blocking.
                    .whenComplete((r, e) -> completed.increment()));
        }

        // 3. Aggregate.
        CompletableFuture<Void> all =
                CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

        return all
                // 4. Collect and summarise.
                .thenApply(v -> {
                    int ok = 0, ko = 0;
                    for (CompletableFuture<NoticeResult> f : futures) {
                        // join() does not block: allOf guarantees they are complete.
                        if (f.join().successful()) ok++; else ko++;
                    }
                    return new SendSummary(total, ok, ko,
                            (System.nanoTime() - start) / 1_000_000);
                })
                // 5. Global deadline and final handler.
                .orTimeout(60, TimeUnit.SECONDS)
                .exceptionally(e -> {
                    System.err.println("Bulk send failed: " + causeOf(e).getMessage());
                    return new SendSummary(total, 0, total,
                            (System.nanoTime() - start) / 1_000_000);
                });
    }

    public long completed() { return completed.sum(); }

    static Throwable causeOf(Throwable t) {
        return (t instanceof CompletionException || t instanceof ExecutionException)
                && t.getCause() != null ? t.getCause() : t;
    }

    @Override public void close() {
        ioPool.shutdown();
        try {
            if (!ioPool.awaitTermination(30, TimeUnit.SECONDS)) ioPool.shutdownNow();
        } catch (InterruptedException e) {
            ioPool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) throws Exception {

        List<Notice> notices = new ArrayList<>();
        for (int i = 1; i <= 200; i++) {
            notices.add(new Notice("NT-" + i, "employee-" + (i % 20)));
        }

        try (AsyncNoticeSending service = new AsyncNoticeSending()) {

            System.out.println("=== ASYNCHRONOUS SENDING OF 200 NOTICES ===");
            System.out.println("Sequential would be " + (200 * 250 / 1000) + " s\n");

            // 6. The chain is declared and returns: main does NOT block.
            CompletableFuture<SendSummary> chain = service.sendAll(notices);

            // Progress, reading the LongAdder the tasks increment.
            while (!chain.isDone()) {
                long done = service.completed();
                System.out.printf("\r  [%s] %d/200 (%.0f%%)",
                        bar(done, 200), done, 100.0 * done / 200);
                TimeUnit.MILLISECONDS.sleep(250);
            }

            SendSummary r = chain.get();
            System.out.printf("\r  [%s] 200/200 (100%%)%n%n", bar(200, 200));
            System.out.println("=== SUMMARY ===");
            System.out.println("Total     : " + r.total());
            System.out.println("Sent      : " + r.sent());
            System.out.println("Failed    : " + r.failed());
            System.out.println("Time      : " + r.ms() + " ms");
            System.out.printf ("Success   : %.1f%%%n", r.percentage());
            System.out.printf ("Speed-up  : %.1fx%n", 200 * 250.0 / r.ms());
        }
    }

    static String bar(long done, long total) {
        int width = 30;
        int filled = (int) (width * done / total);
        return "#".repeat(filled) + "-".repeat(width - filled);
    }
}

=== ASYNCHRONOUS SENDING OF 200 NOTICES ===
Sequential would be 50 s

  [##############################] 200/200 (100%)

=== SUMMARY ===
Total     : 200
Sent      : 193
Failed    : 7
Time      : 3387 ms
Success   : 96.5%
Speed-up  : 14.8x