Overview

CompletableFuture, introduced in JDK 8, represents a significant advancement in Java's asynchronous programming capabilities. This class implements both the CompletionStage and Future interfaces, providing a powerful framework for handling asynchronous operations with enhanced flexibility.

The CompletionStage interface significantly extends Future by introducing callback mechanisms, stream-style processing, and the ability to compose multiple futures together. The abundance of functional programming constructs in CompletionStage's methods demonstrates Java 8's functional programming capabilities.

CompletableFuture serves three primary purposes:

• Creating asynchronous tasks
• Handling simple asynchronous callbacks
• Combining multiple tasks for complex workflows

---

Creating CompletableFuture Instances

Using the new Keyword

The most straightforward approach involves instantiating CompletableFuture directly. This pattern treats CompletableFuture as a basic Future implementation.

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

// Later, when the result becomes available
asyncResult.complete(serviceResponse);

// Retrieve the result - blocks until completion
String result = asyncResult.get();

The complete() method can only be invoked once; subsequent calls are ignored.

Static Factory Methods

For scenarios where the result is already known, the completedFuture() method provides a convenient way to create a pre-completed CompletableFuture:

CompletableFuture<String> future = CompletableFuture.completedFuture("initial-data");
assertEquals("initial-data", future.get());

The internal implementation wraps the value in a new CompletableFuture instance.

---

Asynchronous Task Creation

Two primary methods exist for creating asynchronous tasks: supplyAsync and runAsync.

supplyAsync Method

This method creates a task that returns a result. Two overloaded versions are available:

// Uses default thread pool (ForkJoinPool.commonPool())
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)

// Allows custom thread pool configuration
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

Practical implementation example:

public class AsyncTaskDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        
        CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> {
            System.out.println("Processing in: " + Thread.currentThread().getName());
            return "processed-data";
        }, threadPool);
        
        System.out.println("Result: " + task.get());
        threadPool.shutdown();
    }
}

runAsync Method

Use this method when the task doesn't require returning a value:

// Uses default thread pool
public static CompletableFuture<Void> runAsync(Runnable runnable)

// Allows custom thread pool
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

Complete example demonstrating both methods:

public class TaskDemo {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        
        CompletableFuture<Void> runTask = CompletableFuture.runAsync(
            () -> System.out.println("Running async task"), 
            executor
        );
        
        CompletableFuture<String> supplyTask = CompletableFuture.supplyAsync(() -> {
            System.out.println("Supplying data");
            return "supplied-value";
        }, executor);
        
        System.out.println("Run result: " + runTask.join());
        System.out.println("Supply result: " + supplyTask.join());
        executor.shutdown();
    }
}

Retrieving Task Results

Several methods exist for retrieving results from CompletableFuture:

// Blocks until result is available, throws exception on failure
public T get() throws InterruptedException, ExecutionException

// Blocks with timeout
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException

// Similar to get() but throws unchecked exception on failure
public T join()

// Returns value immediately or given default if not complete
public T getNow(T valueIfAbsent)

// Manually complete the future with a value
public boolean complete(T value)

// Manually complete with an exception
public boolean completeExceptionally(Throwable ex)

---

Asynchronous Callback Processing

thenApply and thenApplyAsync

thenApply processes the result of a previous task and returns a transformed value:

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> initial = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread() + " performing initial computation");
        return 10;
    });

    CompletableFuture<Integer> transformed = initial.thenApplyAsync((value) -> {
        System.out.println(Thread.currentThread() + " transforming value");
        return value * 2;
    });

    System.out.println("Initial result: " + initial.get());
    System.out.println("Transformed result: " + transformed.get());
}

// Output:
// Initial result: 10
// Transformed result: 20

thenAccept and thenAcceptAsync

thenAccept consumes the result without returning a value:

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> initial = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread() + " executing task");
        return 42;
    });

    CompletableFuture<Void> consumer = initial.thenAccept((result) -> {
        System.out.println(Thread.currentThread() + " consuming result: " + result);
    });

    System.out.println("Initial: " + initial.get());
    System.out.println("Consumer: " + consumer.get());
}
// Output:
// Initial: 42
// Consumer: null

thenRun and thenRunAsync

thenRun executes a Runnable after completion, without access to the result:

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> initial = CompletableFuture.supplyAsync(() -> {
        return 100;
    });

    CompletableFuture<Void> followUp = initial.thenRun(() -> {
        System.out.println("Task completed, running follow-up action");
    });

    System.out.println("Initial: " + initial.get());
    System.out.println("Follow-up: " + followUp.get());
}
// Output:
// Initial: 100
// Follow-up: null

whenComplete and whenCompleteAsync

This callback receives both the result and any expection that occurred:

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> initial = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread() + " executing computation");
        int result = 50 / 2;  // Normal execution
        return result;
    });

    CompletableFuture<Integer> completionHandler = initial.whenComplete((result, error) -> {
        System.out.println("Previous result: " + result);
        System.out.println("Exception occurred: " + error);
        System.out.println(Thread.currentThread() + " handling completion");
    });

    System.out.println("Final result: " + completionHandler.get());
}

handle and handleAsync

Similar to whenComplete, but the handler can return a value:

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Integer> initial = CompletableFuture.supplyAsync(() -> {
        System.out.println(Thread.currentThread() + " executing initial task");
        return 15;
    });

    CompletableFuture<Integer> handled = initial.handle((result, error) -> {
        System.out.println(Thread.currentThread() + " handling result");
        System.out.println("Result: " + result);
        System.out.println("Error: " + error);
        return result != null ? result + 5 : 0;
    });

    System.out.println("Handled result: " + handled.get());
}

// Output:
// Handled result: 20

Synchronous vs Asynchronous Variants

The distinction between then*** and then***Async methods is crucial:

• then* methods execute using the same thread as the previous task
• then*Async methods spawn a new thread, optionally from a custom executor
• Async variants default to ForkJoinPool.commonPool() when no executor is specified

---

Combining Multiple Tasks

AND Composition

The thenCombine, thenAcceptBoth, and runAfterBoth methods coordinate two futures, executing the next stage only when both complete successfully.

public class CombineDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> first = CompletableFuture.completedFuture("task-one");
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        CompletableFuture<String> combined = CompletableFuture
            .supplyAsync(() -> "task-two", executor)
            .thenCombineAsync(first, (secondResult, firstResult) -> {
                System.out.println("First: " + firstResult);
                System.out.println("Second: " + secondResult);
                return "combined-result";
            }, executor);
            
        System.out.println(combined.join());
        executor.shutdown();
    }
}
// Output:
// First: task-one
// Second: task-two
// combined-result

Key differences:

• thenCombine: Combines results and returns a new value
• thenAcceptBoth: Consumes both results, returns void
• runAfterBoth: Executes after both complete, no access to results

When either input future completes exceptionally, the resulting future is completed exceptionally with that exception.

OR Composition

The applyToEither, acceptEither, and runAfterEither methods trigger when either of two futures completes.

public class EitherDemo {
    public static void main(String[] args) {
        CompletableFuture<String> first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(3000L);
                System.out.println("First task finished");
            } catch (InterruptedException e) {
                return "first-interrupted";
            }
            return "first-result";
        });
        
        ExecutorService executor = Executors.newSingleThreadExecutor();
        CompletableFuture<Void> eitherTask = CompletableFuture
            .supplyAsync(() -> {
                System.out.println("Second task finished");
                return "second-result";
            }, executor)
            .acceptEitherAsync(first, System.out::println, executor);
            
        eitherTask.join();
        executor.shutdown();
    }
}
// Output:
// Second task finished
// second-result

Distinctions:

• applyToEither: Uses completed result as parameter, returns a value
• acceptEither: Uses completed result as parameter, returns void
• runAfterEither: Executes regardless of result, no parameter access

AllOf - Waiting for All Futures

allOf creates a CompletableFuture that completes when all input futures complete:

public class AllOfDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Void> taskA = CompletableFuture.runAsync(() -> 
            System.out.println("Task A completed")
        );
        CompletableFuture<Void> taskB = CompletableFuture.runAsync(() -> 
            System.out.println("Task B completed")
        );
        
        CompletableFuture<Void> allDone = CompletableFuture.allOf(taskA, taskB)
            .whenComplete((result, error) -> 
                System.out.println("All tasks finished")
            );
            
        allDone.join();
    }
}
// Output:
// Task A completed
// Task B completed
// All tasks finished

If any future throws an exception, calling get() on the allOf result throws an ExecutionException.

AnyOf - Waiting for First Future

anyOf completes when any input future completes:

public class AnyOfDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Void> slowTask = CompletableFuture.runAsync(() -> {
            try {
                Thread.sleep(2000L);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println("Slow task completed");
        });
        
        CompletableFuture<Void> fastTask = CompletableFuture.runAsync(() -> 
            System.out.println("Fast task completed")
        );
        
        CompletableFuture<Object> firstComplete = CompletableFuture.anyOf(slowTask, fastTask)
            .whenComplete((result, error) -> 
                System.out.println("First task finished, others may continue")
            );
            
        firstComplete.join();
    }
}
// Output:
// Fast task completed
// First task finished, others may continue

thenCompose for Sequential Execution

thenCompose flattens nested CompletableFutures, enabling sequential async operations:

public class ComposeDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> baseTask = CompletableFuture.completedFuture("base-data");
        ExecutorService executor = Executors.newSingleThreadExecutor();
        
        CompletableFuture<String> chained = CompletableFuture
            .supplyAsync(() -> "step-one", executor)
            .thenComposeAsync(result -> {
                System.out.println("Step one produced: " + result);
                return baseTask;  // Returns a new CompletableFuture
            }, executor);
            
        System.out.println("Final result: " + chained.join());
        executor.shutdown();
    }
}
// Output:
// Step one produced: step-one
// Final result: base-data

---

Practical Considerations

Exception Handling Requirements

Calling get() or join() is necessary to observe exceptions; otherwise, failures go undetected:

ExecutorService executor = new ThreadPoolExecutor(4, 8, 30L,
    TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));

CompletableFuture<Boolean> task = CompletableFuture.supplyAsync(() -> {
    int result = 10 / 0;  // ArithmeticException
    return true;
}, executor).thenAccept(System.out::println);

// Without get() or join(), the exception remains hidden
task.join();

Blocking Behavior

The get() method blocks indefinitely. Always specify a timeout:

// Avoid - potential indefinite blocking
future.get();

// Recommended - with timeout
future.get(5, TimeUnit.SECONDS);

Prefer join() for cleaner code when exception handling isn't critical, as it doesn't require explicit exception declarations.

Default Thread Pool Limitations

CompletableFuture's default thread pool (ForkJoinPool.commonPool()) uses CPU core count minus one thread. Under high load with complex processing, this creates a bottleneck. Custom thread pools with appropriate sizing are recommended:

ExecutorService customPool = new ThreadPoolExecutor(
    10, 20, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(500),
    new ThreadPoolExecutor.CallerRunsPolicy()
);

Thread Pool Saturation Policies

When using custom thread pools, avoid DiscardPolicy or DiscardOldestPolicy with CompletableFuture. These policies silently drop tasks without signaling failure. Use AbortPolicy and implement proper timeout handling:

ExecutorService executor = new ThreadPoolExecutor(
    10, 20, 60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(),
    new ThreadPoolExecutor.AbortPolicy()
);

CompletableFuture.runAsync(() -> processData(), executor)
    .orTimeout(5, TimeUnit.SECONDS)
    .whenComplete((result, error) -> {
        if (error != null) {
            log.error("Task failed", error);
        }
    });

Consider isolating long-running tasks into separate thread pools to prevent resource contention.