CompletableFuture in Java — Why Silent Failures Corrupt

import java.util.concurrent.*;

public class OldVsNewAsync {

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

        ExecutorService executor = Executors.newFixedThreadPool(2);

        // ─── THE OLD WAY (Future) ───────────────────────────────────────
        // Submit a slow task to a thread pool
        Future<String> oldStyleFuture = executor.submit(() -> {
            Thread.sleep(1000); // Simulates a slow database call
            return "User data from database";
        });

        // This line BLOCKS — the main thread freezes here until the result arrives
        // You cannot attach a callback or chain another step
        String oldResult = oldStyleFuture.get(); // Thread stuck waiting here
        System.out.println("Old way result: " + oldResult);

        // ─── THE NEW WAY (CompletableFuture) ────────────────────────────
        // supplyAsync kicks off the task on a background thread and returns immediately
        CompletableFuture<String> modernFuture = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000); // Simulates a slow API call
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Order details from payment API";
        }, executor);

        // thenApply attaches a callback — this runs automatically when the result arrives
        // The main thread is NOT blocked — it's free to do other work right now
        CompletableFuture<String> processedFuture = modernFuture.thenApply(rawResult -> {
            // Transform the result — like a pipeline stage
            return "Processed: " + rawResult.toUpperCase();
        });

        // We only block at the very end when we truly need the final answer
        System.out.println("New way result: " + processedFuture.get());

        executor.shutdown();
    }
}

import java.util.concurrent.*;

public class OrderPipelineExample {

    // Simulates fetching a raw order ID from a slow database
    static String fetchOrderId() {
        try { Thread.sleep(500); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return "ORD-78421";
    }

    // Simulates enriching the order with product details from another service
    static String enrichWithProductDetails(String orderId) {
        try { Thread.sleep(300); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return orderId + " | Product: Wireless Headphones | Qty: 1";
    }

    // Simulates applying a discount rule
    static String applyLoyaltyDiscount(String enrichedOrder) {
        return enrichedOrder + " | Discount: 10% applied";
    }

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

        // Use a named thread pool — always do this in real code
        ExecutorService orderProcessingPool = Executors.newFixedThreadPool(4);

        long startTime = System.currentTimeMillis();

        CompletableFuture<Void> orderPipeline = CompletableFuture

            // STEP 1: Start the async task — kicks off on a background thread
            // supplyAsync() returns a CompletableFuture<String>
            .supplyAsync(OrderPipelineExample::fetchOrderId, orderProcessingPool)

            // STEP 2: Transform the result — thenApply() is like .map() in streams
            // Runs automatically when step 1 finishes, on a pool thread
            .thenApply(OrderPipelineExample::enrichWithProductDetails)

            // STEP 3: Another transformation — chain keeps flowing
            .thenApply(OrderPipelineExample::applyLoyaltyDiscount)

            // STEP 4: thenAccept() is the final consumer — returns CompletableFuture<Void>
            // Use this when you want to DO something with the result, not transform it
            .thenAccept(finalOrderSummary -> {
                System.out.println("✅ Order ready to dispatch:");
                System.out.println("   " + finalOrderSummary);
            });

        // The main thread has been free this whole time
        // We wait here only because we need the pipeline to finish before printing the time
        orderPipeline.get(); // Blocks only at the very end

        long elapsed = System.currentTimeMillis() - startTime;
        System.out.println("⏱  Total time: " + elapsed + "ms (sequential steps, ~800ms expected)");

        orderProcessingPool.shutdown();
    }
}

import java.util.concurrent.*;
import java.util.List;

public class ParallelDashboardLoader {

    // Three independent data sources — each takes different time to respond
    static String fetchUserProfile(String userId) {
        try { Thread.sleep(400); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return "Profile[name=Alice, tier=Gold]";
    }

    static String fetchRecentOrders(String userId) {
        try { Thread.sleep(600); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return "Orders[ORD-001, ORD-002, ORD-003]";
    }

    static Integer fetchLoyaltyPoints(String userId) {
        try { Thread.sleep(300); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return 2850;
    }

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

        ExecutorService dashboardPool = Executors.newFixedThreadPool(6);
        String userId = "user-4492";

        long startTime = System.currentTimeMillis();

        // ─── LAUNCH ALL THREE TASKS IN PARALLEL ─────────────────────────
        // None of these block — they all start immediately on background threads
        CompletableFuture<String> profileFuture =
            CompletableFuture.supplyAsync(() -> fetchUserProfile(userId), dashboardPool);

        CompletableFuture<String> ordersFuture =
            CompletableFuture.supplyAsync(() -> fetchRecentOrders(userId), dashboardPool);

        CompletableFuture<Integer> pointsFuture =
            CompletableFuture.supplyAsync(() -> fetchLoyaltyPoints(userId), dashboardPool);

        // ─── thenCombine: merge two futures when BOTH are done ───────────
        // profileFuture and ordersFuture run in parallel
        // thenCombine fires only after BOTH complete
        CompletableFuture<String> profileAndOrders = profileFuture.thenCombine(
            ordersFuture,
            (profile, orders) -> profile + " | " + orders  // BiFunction merging both results
        );

        // ─── allOf: wait for ALL futures in parallel ─────────────────────
        // Pass all three futures — allOf itself returns CompletableFuture<Void>
        CompletableFuture<Void> allDataLoaded =
            CompletableFuture.allOf(profileFuture, ordersFuture, pointsFuture);

        // Block until all three are done — then read each result with .get()
        // At this point all futures are already complete, so .get() returns instantly
        allDataLoaded.get();

        long elapsed = System.currentTimeMillis() - startTime;

        // Assemble the dashboard from all three parallel results
        String dashboardSummary = String.format(
            "Dashboard for %s:\n  %s\n  %s\n  Loyalty Points: %d",
            userId,
            profileFuture.get(),
            ordersFuture.get(),
            pointsFuture.get()
        );

        System.out.println(dashboardSummary);
        System.out.println("⏱  Total time: " + elapsed +
            "ms (parallel — slowest task was 600ms, NOT 1300ms sequential)");

        dashboardPool.shutdown();
    }
}

import java.util.concurrent.*;

public class ResilientPaymentProcessor {

    // Simulates a payment gateway that sometimes fails
    static String chargePaymentGateway(double amount) {
        if (amount > 10000) {
            // Simulate a gateway rejection for large amounts
            throw new RuntimeException("Payment gateway rejected: amount exceeds daily limit");
        }
        return "PAYMENT_SUCCESS | TxnId: TXN-" + (int)(Math.random() * 99999) + " | Amount: $" + amount;
    }

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

        ExecutorService paymentPool = Executors.newFixedThreadPool(2);

        // ─── SCENARIO 1: Successful payment ─────────────────────────────
        System.out.println("=== Scenario 1: Normal payment ===");

        String successResult = CompletableFuture
            .supplyAsync(() -> chargePaymentGateway(250.00), paymentPool)

            // exceptionally() only fires if an exception occurred upstream
            // It receives the exception and must return a fallback value of the same type
            .exceptionally(exception -> {
                System.out.println("  ⚠ Payment failed: " + exception.getMessage());
                return "PAYMENT_FAILED | Fallback: queued for retry";
            })
            .get();

        System.out.println("  Result: " + successResult);

        // ─── SCENARIO 2: Failed payment with fallback ────────────────────
        System.out.println("\n=== Scenario 2: Oversized payment ===");

        String failureResult = CompletableFuture
            .supplyAsync(() -> chargePaymentGateway(15000.00), paymentPool) // This will throw
            .exceptionally(exception -> {
                // exception here is a CompletionException wrapping the real cause
                Throwable rootCause = exception.getCause(); // Always unwrap to get the real error
                System.out.println("  ⚠ Caught error: " + rootCause.getMessage());
                return "PAYMENT_FAILED | Fallback: queued for retry";
            })
            .get();

        System.out.println("  Result: " + failureResult);

        // ─── SCENARIO 3: handle() — runs in BOTH success and failure cases
        System.out.println("\n=== Scenario 3: Using handle() for unified success/failure logic ===");

        String handleResult = CompletableFuture
            .supplyAsync(() -> chargePaymentGateway(500.00), paymentPool)

            // handle() gives you BOTH the result and the exception
            // Exactly one of them will be null — the other will have a value
            .handle((paymentConfirmation, exception) -> {
                if (exception != null) {
                    // Task failed — exception has the error, paymentConfirmation is null
                    return "HANDLED_FAILURE: " + exception.getCause().getMessage();
                } else {
                    // Task succeeded — paymentConfirmation has the value, exception is null
                    return "HANDLED_SUCCESS: " + paymentConfirmation;
                }
            })
            .get();

        System.out.println("  Result: " + handleResult);

        paymentPool.shutdown();
    }
}

import java.util.concurrent.*;
import java.util.function.BiFunction;

public class UserDashboardService {

    // Simulated external API calls
    static class UserProfile { String name; int loyaltyPoints; /* getters & setters omitted */ public UserProfile(String name, int points) { this.name = name; this.loyaltyPoints = points; } }
    static class OrderHistory { int orderCount; String lastOrder; /* getters omitted */ public OrderHistory(int count, String last) { this.orderCount = count; this.lastOrder = last; } }
    static class DashboardDto { UserProfile profile; OrderHistory orders; /* constructor, getters omitted */ public DashboardDto(UserProfile p, OrderHistory o) { profile = p; orders = o; } }

    private final ExecutorService asyncPool = Executors.newFixedThreadPool(10);

    public CompletableFuture<DashboardDto> loadDashboardAsync(String userId) {

        // Fire both API calls in parallel immediately
        CompletableFuture<UserProfile> profileFuture = CompletableFuture.supplyAsync(() -> {
            // Simulate remote call
            try { Thread.sleep(400); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            return new UserProfile("Alice", 3500);
        }, asyncPool).exceptionally(ex -> {
            // Log error and return fallback
            System.err.println("Profile service failed: " + ex.getCause().getMessage());
            return new UserProfile("Unknown", 0);
        });

        CompletableFuture<OrderHistory> ordersFuture = CompletableFuture.supplyAsync(() -> {
            try { Thread.sleep(600); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            return new OrderHistory(12, "ORD-99812");
        }, asyncPool).exceptionally(ex -> {
            System.err.println("Order service failed: " + ex.getCause().getMessage());
            return new OrderHistory(0, "None");
        });

        // Combine both results when both are ready
        return profileFuture.thenCombine(ordersFuture,
            (profile, orders) -> new DashboardDto(profile, orders));
    }

    // Shutdown pool gracefully when application stops
    public void shutdown() { asyncPool.shutdown(); }
}

// io.thecodeforge
import java.util.concurrent.*;

public class TimeoutGuard {
    private static final ExecutorService pool = Executors.newFixedThreadPool(10);

    public CompletableFuture<String> fetchWithFallback(String query) {
        return CompletableFuture.supplyAsync(() -> {
            // Simulate a slow downstream call
            sleep(2000);
            return "live data for " + query;
        }, pool)
        .completeOnTimeout("cached data for " + query, 1, TimeUnit.SECONDS)
        .exceptionally(ex -> "fallback: degraded mode");
    }

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

// io.thecodeforge
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.*;

@Configuration
public class PoolConfiguration {

    @Bean
    public Executor paymentGatewayPool() {
        return new ThreadPoolExecutor(
            5, 5,        // core, max
            30, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(100),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    @Bean
    public Executor imageProcessingPool() {
        return new ThreadPoolExecutor(
            2, 2,
            30, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(500),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }
}