// Old Java Future: blocking, no composition
Future<String> future = executorService.submit(() -> callExternalApi());
String result = future.get();  // BLOCKS the calling thread!
 
// CompletableFuture: non-blocking, chainable
CompletableFuture.supplyAsync(() -> callExternalApi())
    .thenApply(result -> transformResult(result))
    .thenAccept(result -> saveToDatabase(result))
    .exceptionally(ex -> { handleError(ex); return null; });
// The calling thread returns IMMEDIATELY. Callbacks run on worker threads.

// Method 1: supplyAsync - runs a Supplier<T>, produces a result
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "Hello from async";  // Returns a value
});
 
// Method 2: runAsync - runs a Runnable, no result
CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
    sendEmail("welcome@example.com");  // No return value
});
 
// Method 3: completedFuture - already-completed future (useful in tests/fallbacks)
CompletableFuture<String> alreadyDone = CompletableFuture.completedFuture("cached-value");
 
// Method 4: failedFuture - already-failed future (useful in tests)
CompletableFuture<String> alreadyFailed = CompletableFuture.failedFuture(
    new RuntimeException("Simulated failure")
);

ExecutorService myPool = Executors.newFixedThreadPool(20);
 
// supplyAsync with custom executor
CompletableFuture<User> future = CompletableFuture.supplyAsync(
    () -> userRepository.findById(userId),
    myPool  // Use THIS pool, not the default ForkJoinPool
);

CompletableFuture<String> promise = new CompletableFuture<>();
 
// Pass `promise` to a callback-based API
legacyCallbackApi.fetch(result -> {
    promise.complete(result);        // Normal completion
}, error -> {
    promise.completeExceptionally(error);  // Exceptional completion
});
 
// Caller waits non-blockingly via chaining
promise.thenAccept(result -> process(result));

CompletableFuture<String> rawResult = CompletableFuture.supplyAsync(() -> fetchRawData());
 
// thenApply: transform the result (String -> Integer)
// Equivalent to map() in streams
CompletableFuture<Integer> length = rawResult.thenApply(String::length);
 
// thenAccept: consume the result, return void
// Equivalent to forEach in streams
CompletableFuture<Void> consumed = rawResult.thenAccept(result -> {
    database.save(result);
});
 
// thenRun: execute something after completion (ignores the result)
CompletableFuture<Void> afterRun = rawResult.thenRun(() -> {
    metrics.incrementCounter("fetch.completed");
});

// thenApply (no Async): callback runs on the SAME thread that completed the future
// (or the calling thread if already complete)
future.thenApply(x -> transform(x));
 
// thenApplyAsync: callback always runs on a DIFFERENT thread (from ForkJoinPool or your executor)
future.thenApplyAsync(x -> transform(x));
future.thenApplyAsync(x -> transform(x), myExecutor);  // Use specific pool

// Use thenApply (no Async) when:
// - The transformation is fast and non-blocking (pure computation)
// - You want to avoid thread context switches
future.thenApply(result -> result.toUpperCase())  // Fast string operation: no Async needed
 
// Use thenApplyAsync when:
// - The transformation does I/O (database, HTTP call)
// - The transformation is slow (avoid blocking the completing thread)
future.thenApplyAsync(result -> {
    return userRepository.findByEmail(result);  // I/O: use Async
}, dbExecutor);

// WRONG: using thenApply with an async operation returns a nested future
CompletableFuture<CompletableFuture<User>> nested =
    CompletableFuture.supplyAsync(() -> userId)
        .thenApply(id -> fetchUserAsync(id));  // Returns CF<CF<User>> - wrong!
 
// CORRECT: thenCompose flattens the nested future
CompletableFuture<User> flat =
    CompletableFuture.supplyAsync(() -> userId)
        .thenCompose(id -> fetchUserAsync(id));  // Returns CF<User> - correct!
 
// Real-world example: order placement pipeline
CompletableFuture<OrderConfirmation> placeOrder(OrderRequest request) {
    return validateUser(request.getUserId())           // CF<UserValidation>
        .thenCompose(validation ->
            reserveInventory(request.getItems())       // CF<Reservation>
        )
        .thenCompose(reservation ->
            processPayment(request.getPaymentInfo())   // CF<PaymentResult>
        )
        .thenCompose(payment ->
            scheduleDelivery(payment.getOrderId())     // CF<DeliverySchedule>
        )
        .thenApply(delivery ->
            buildConfirmation(delivery)                // Synchronous transform
        );
}
// Each step waits for the previous step. Sequential pipeline.

// Run both in parallel, then merge results when BOTH complete
CompletableFuture<UserProfile> userFuture =
    CompletableFuture.supplyAsync(() -> userService.getProfile(userId));
 
CompletableFuture<List<Order>> ordersFuture =
    CompletableFuture.supplyAsync(() -> orderService.getRecentOrders(userId));
 
// thenCombine: when both complete, merge them
CompletableFuture<DashboardData> dashboard =
    userFuture.thenCombine(ordersFuture, (profile, orders) ->
        new DashboardData(profile, orders)
    );
 
// Real timing benefit:
// Without parallel: 100ms (user) + 150ms (orders) = 250ms total
// With thenCombine: max(100ms, 150ms) = 150ms total  <- 40% faster

/**
 * Classic use case: enriching an entity with data from multiple services.
 * All calls happen in parallel. We wait for all of them.
 */
@Service
@Slf4j
public class ProductEnrichmentService {
 
    public EnrichedProduct enrichProduct(Long productId) {
        // All three calls start SIMULTANEOUSLY
        CompletableFuture<ProductDetails> detailsFuture =
            CompletableFuture.supplyAsync(
                () -> productRepository.findById(productId),
                ioExecutor
            );
 
        CompletableFuture<List<Review>> reviewsFuture =
            CompletableFuture.supplyAsync(
                () -> reviewClient.getReviews(productId),
                ioExecutor
            );
 
        CompletableFuture<InventoryStatus> inventoryFuture =
            CompletableFuture.supplyAsync(
                () -> inventoryClient.getStatus(productId),
                ioExecutor
            );
 
        CompletableFuture<PricingInfo> pricingFuture =
            CompletableFuture.supplyAsync(
                () -> pricingService.getPrice(productId),
                ioExecutor
            );
 
        // Wait for ALL four to complete
        CompletableFuture<Void> allDone =
            CompletableFuture.allOf(detailsFuture, reviewsFuture, inventoryFuture, pricingFuture);
 
        // When all complete, build the enriched product
        return allDone.thenApply(v -> EnrichedProduct.builder()
            .details(detailsFuture.join())    // .join() does not throw checked exceptions
            .reviews(reviewsFuture.join())    // Safe to call after allOf completes
            .inventory(inventoryFuture.join())
            .pricing(pricingFuture.join())
            .build()
        ).join();
 
        // Total time: max(details, reviews, inventory, pricing) - NOT their sum
        // If each takes 100ms, total is 100ms (not 400ms)
    }
}

/**
 * Resilient allOf: if some futures fail, return partial results.
 * Do not fail the entire operation because one enrichment source is down.
 */
public EnrichedProduct enrichProductResilient(Long productId) {
    CompletableFuture<ProductDetails> detailsFuture = fetchDetails(productId)
        .exceptionally(ex -> {
            log.warn("Failed to fetch details for {}: {}", productId, ex.getMessage());
            return ProductDetails.empty();  // Fallback value
        });
 
    CompletableFuture<List<Review>> reviewsFuture = fetchReviews(productId)
        .exceptionally(ex -> {
            log.warn("Failed to fetch reviews for {}: {}", productId, ex.getMessage());
            return List.of();  // Empty list fallback
        });
 
    CompletableFuture<PricingInfo> pricingFuture = fetchPricing(productId)
        .exceptionally(ex -> {
            log.error("CRITICAL: Pricing service down for {}", productId);
            throw new ProductUnavailableException(productId);  // Propagate critical failures
        });
 
    return CompletableFuture.allOf(detailsFuture, reviewsFuture, pricingFuture)
        .thenApply(v -> EnrichedProduct.builder()
            .details(detailsFuture.join())
            .reviews(reviewsFuture.join())
            .pricing(pricingFuture.join())
            .build()
        ).join();
}

/**
 * anyOf: complete as soon as the FIRST future completes.
 *
 * Use case: call multiple data sources, use whichever responds first.
 * Example: Query primary DB and replica simultaneously, return first response.
 * Example: Try multiple CDN nodes, use first to respond.
 */
public User getUserFast(Long userId) {
    CompletableFuture<User> primaryDbFuture =
        CompletableFuture.supplyAsync(() -> primaryDb.findUser(userId), ioExecutor);
 
    CompletableFuture<User> replicaDbFuture =
        CompletableFuture.supplyAsync(() -> replicaDb.findUser(userId), ioExecutor);
 
    CompletableFuture<User> cacheFuture =
        CompletableFuture.supplyAsync(() -> cacheService.getUser(userId), ioExecutor);
 
    // Return whichever completes first
    return (User) CompletableFuture.anyOf(cacheFuture, replicaDbFuture, primaryDbFuture)
        .join();
 
    // NOTE: The other two futures continue running in the background.
    // Cancel them if you want to avoid waste (see below).
}

CompletableFuture<User> userFuture =
    CompletableFuture.supplyAsync(() -> userService.findById(userId))
        .exceptionally(ex -> {
            // Called ONLY if the upstream stage failed
            log.warn("User lookup failed, using default: {}", ex.getMessage());
            return User.anonymous();  // Provide a fallback value
        });
// If the upstream succeeded: exceptionally is SKIPPED
// If the upstream failed: exceptionally provides the fallback

CompletableFuture<Response> responseFuture =
    CompletableFuture.supplyAsync(() -> callExternalApi())
        .handle((result, ex) -> {
            // Called for BOTH success and failure
            if (ex != null) {
                log.error("API call failed: {}", ex.getMessage());
                return Response.error("API unavailable");
            }
            return Response.success(result);
        });
// Unlike exceptionally, handle always runs and can transform the success result too

CompletableFuture<Order> orderFuture =
    CompletableFuture.supplyAsync(() -> createOrder(request))
        .whenComplete((order, ex) -> {
            // Always called. For side effects (logging, metrics).
            // Does NOT change the result.
            if (ex != null) {
                metrics.incrementCounter("order.create.failed");
                log.error("Order creation failed: {}", ex.getMessage());
            } else {
                metrics.incrementCounter("order.create.success");
                log.info("Order created: {}", order.getId());
            }
        });
// The future's value/exception is UNCHANGED by whenComplete

CompletableFuture<Product> result =
    CompletableFuture.supplyAsync(() -> productDb.find(id))
        .exceptionally(ex -> {
            // First recovery attempt: try cache
            log.warn("DB failed, trying cache");
            return productCache.get(id);  // May also throw
        })
        .exceptionally(ex -> {
            // Second recovery: return empty product
            log.error("Both DB and cache failed");
            return Product.empty();
        })
        .thenApply(product -> {
            // Only runs if we have a non-null product (from DB, cache, or empty)
            return enrichProduct(product);
        });

// When you call supplyAsync() WITHOUT an executor, it uses ForkJoinPool.commonPool()
CompletableFuture.supplyAsync(() -> userRepository.findById(id));  // Uses commonPool!
 
// ForkJoinPool.commonPool() has:
// - Threads: CPU_cores - 1 (on a 4-core machine: 3 threads)
// - Designed for: CPU-bound work-stealing tasks
// - NOT designed for: I/O blocking tasks
 
// What happens with I/O on ForkJoinPool:
// Request 1: Blocks on DB call -> occupies 1 of 3 threads
// Request 2: Blocks on DB call -> occupies 1 of 3 threads
// Request 3: Blocks on DB call -> occupies 1 of 3 threads
// Request 4: Waits in queue because all 3 threads are blocked on I/O
//
// ENTIRE APPLICATION STALLS with just 3 simultaneous slow DB queries!

@Configuration
public class ExecutorConfig {
 
    /**
     * For I/O-bound operations: database calls, HTTP calls, file reads.
     * Many threads: threads spend most time waiting, so more threads = more parallelism.
     * Rule of thumb: (threads = target_utilization / (1 - blocking_ratio))
     * For 80% blocking ratio, target 5x CPU cores.
     */
    @Bean("ioExecutor")
    public ExecutorService ioExecutor() {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
            cpuCores * 5,    // Core pool size: always available
            cpuCores * 20,   // Max pool size: burst capacity
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(500),  // Queue for burst protection
            new ThreadFactoryBuilder()
                .setNameFormat("io-async-%d")
                .setDaemon(true)
                .build(),
            new ThreadPoolExecutor.CallerRunsPolicy()  // Don't drop tasks; back-pressure to caller
        );
        return pool;
    }
 
    /**
     * For CPU-bound operations: computation, transformations, encoding.
     * Fewer threads: more threads than CPUs means context switching overhead.
     * Rule of thumb: CPU cores + 1
     */
    @Bean("cpuExecutor")
    public ExecutorService cpuExecutor() {
        int cpuCores = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(
            cpuCores + 1,
            cpuCores + 1,   // Fixed: no benefit from more threads than CPUs for CPU work
            0L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(200),
            new ThreadFactoryBuilder().setNameFormat("cpu-async-%d").setDaemon(true).build()
        );
    }
 
    /**
     * For external HTTP calls to third-party services.
     * Separate from DB I/O: one slow external API should not affect DB query throughput.
     */
    @Bean("httpCallExecutor")
    public ExecutorService httpCallExecutor() {
        return new ThreadPoolExecutor(
            20, 100, 30L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(300),
            new ThreadFactoryBuilder().setNameFormat("http-async-%d").setDaemon(true).build()
        );
    }
 
    /**
     * For scatter-gather: query all database shards in parallel.
     * Fixed size = number of shards (bounded parallelism).
     */
    @Bean("scatterGatherExecutor")
    public ExecutorService scatterGatherExecutor(
            @Value("${app.sharding.num-shards:4}") int numShards) {
        return Executors.newFixedThreadPool(
            numShards,
            new ThreadFactoryBuilder().setNameFormat("scatter-gather-%d").setDaemon(true).build()
        );
    }
}

@Configuration
public class ExecutorMetricsConfig {
 
    /**
     * Register all thread pools with Micrometer for monitoring.
     * Provides metrics like:
     *   executor.pool.size{name="ioExecutor"}    <- Current pool size
     *   executor.queued{name="ioExecutor"}        <- Queued tasks
     *   executor.completed{name="ioExecutor"}     <- Completed tasks
     */
    @Bean
    public ExecutorServiceMetrics ioExecutorMetrics(
            @Qualifier("ioExecutor") ExecutorService ioExecutor,
            MeterRegistry registry) {
        return ExecutorServiceMetrics.monitor(registry, ioExecutor, "ioExecutor");
    }
}

@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {
 
    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("spring-async-");
        executor.initialize();
        return executor;
    }
}
 
@Service
public class ReportService {
 
    /**
     * @Async + CompletableFuture: the method is called synchronously,
     * but Spring AOP intercepts it, submits the actual body to the async executor,
     * and returns the CompletableFuture immediately.
     *
     * The caller gets the CompletableFuture back immediately.
     * The report generation runs in the background.
     * The caller can .get(), .thenApply(), or just ignore it.
     */
    @Async
    public CompletableFuture<Report> generateMonthlyReport(Month month) {
        // This runs on the Spring async executor thread pool
        Report report = reportGenerator.generate(month);
        return CompletableFuture.completedFuture(report);
    }
}
 
// Controller using the async service:
@RestController
public class ReportController {
 
    @GetMapping("/reports/monthly")
    public CompletableFuture<ResponseEntity<Report>> getMonthlyReport(
            @RequestParam Month month) {
 
        return reportService.generateMonthlyReport(month)
            .thenApply(report -> ResponseEntity.ok(report))
            .exceptionally(ex -> ResponseEntity.internalServerError().build());
        // Spring MVC supports CompletableFuture as a return type from controllers.
        // The request thread is released immediately. Response is sent when future completes.
    }
}

CompletableFuture<User> userFuture =
    CompletableFuture.supplyAsync(() -> userService.findById(userId), ioExecutor)
        .orTimeout(3, TimeUnit.SECONDS);  // Throw TimeoutException if not done in 3s
 
// If the future does not complete in 3 seconds:
//   TimeoutException is thrown (wrapped in CompletionException)
// If the future completes before 3 seconds:
//   Behaves normally
 
userFuture.exceptionally(ex -> {
    if (ex.getCause() instanceof TimeoutException) {
        log.warn("User service timed out for userId={}", userId);
        return User.anonymous();  // Fallback on timeout
    }
    throw new RuntimeException(ex);
});

CompletableFuture<String> result =
    CompletableFuture.supplyAsync(() -> slowExternalService.getData(), httpExecutor)
        .completeOnTimeout("default-value", 2, TimeUnit.SECONDS);
// If not done in 2 seconds: completes with "default-value" instead of throwing
// If done before 2 seconds: completes with the actual result

@Service
@Slf4j
public class ProductSearchService {
 
    private static final Duration SEARCH_TIMEOUT = Duration.ofMillis(500);
 
    public SearchResult search(String query) {
        return CompletableFuture
            .supplyAsync(() -> elasticSearchClient.search(query), httpCallExecutor)
            .orTimeout(SEARCH_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)
            .exceptionally(ex -> {
                if (ex.getCause() instanceof TimeoutException) {
                    log.warn("Elasticsearch timeout ({}ms) for query='{}'. Using DB fallback.",
                        SEARCH_TIMEOUT.toMillis(), query);
                    // Fallback: slower but more reliable database search
                    return dbSearchService.search(query);
                }
                log.error("Search failed for query='{}': {}", query, ex.getMessage());
                return SearchResult.empty();
            })
            .join();
    }
}

@Service
@Slf4j
public class ProductPageService {
 
    private final ProductRepository productRepository;
    private final ReviewServiceClient reviewClient;
    private final InventoryClient inventoryClient;
    private final RecommendationClient recommendationClient;
    private final ExecutorService ioExecutor;
 
    /**
     * A product page needs data from 4 independent sources.
     * Without parallelism: 100 + 150 + 80 + 120 = 450ms
     * With allOf parallelism: max(100, 150, 80, 120) = 150ms
     *
     * This is a 3x speedup with no architectural changes.
     */
    public ProductPage getProductPage(Long productId, Long userId) {
        log.debug("Fetching product page for productId={}, userId={}", productId, userId);
 
        CompletableFuture<Product> productFuture =
            CompletableFuture.supplyAsync(
                () -> productRepository.findByIdOrThrow(productId),
                ioExecutor
            );
 
        CompletableFuture<ReviewSummary> reviewsFuture =
            CompletableFuture.supplyAsync(
                () -> reviewClient.getSummary(productId),
                ioExecutor
            )
            .orTimeout(300, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> {
                log.warn("Reviews unavailable for product {}: {}", productId, ex.getMessage());
                return ReviewSummary.empty();  // Non-critical: show page without reviews
            });
 
        CompletableFuture<InventoryInfo> inventoryFuture =
            CompletableFuture.supplyAsync(
                () -> inventoryClient.getInfo(productId),
                ioExecutor
            )
            .orTimeout(200, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> {
                log.warn("Inventory info unavailable for product {}", productId);
                return InventoryInfo.unknown();  // Show "may be available"
            });
 
        CompletableFuture<List<Product>> recommendationsFuture =
            CompletableFuture.supplyAsync(
                () -> recommendationClient.getSimilar(productId, userId, 5),
                ioExecutor
            )
            .orTimeout(400, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> List.of());  // Non-critical: show page without recs
 
        // All four requests are running NOW in parallel
        // Wait for all, then build the page
        try {
            CompletableFuture.allOf(productFuture, reviewsFuture, inventoryFuture, recommendationsFuture)
                .orTimeout(1, TimeUnit.SECONDS)  // Overall page timeout
                .join();
        } catch (CompletionException ex) {
            if (ex.getCause() instanceof TimeoutException) {
                log.error("Product page overall timeout for productId={}", productId);
                // Individual futures have their own fallbacks, so we can still build a partial page
            }
        }
 
        return ProductPage.builder()
            .product(productFuture.getNow(Product.empty()))             // getNow: get or default
            .reviews(reviewsFuture.getNow(ReviewSummary.empty()))
            .inventory(inventoryFuture.getNow(InventoryInfo.unknown()))
            .recommendations(recommendationsFuture.getNow(List.of()))
            .build();
    }
}

@Service
@Slf4j
public class NotificationService {
 
    /**
     * Send the same notification via multiple channels simultaneously.
     * Fire and forget: we do not wait for all channels to complete.
     * If one channel fails, others still succeed.
     */
    public void notifyOrderPlaced(Order order, User user) {
        String correlationId = RequestContext.getCorrelationId();  // Capture before async
 
        // Fire all notification channels in parallel - do not wait
        CompletableFuture.runAsync(() -> {
            MDC.put("correlationId", correlationId);
            try {
                emailService.sendOrderConfirmation(user.getEmail(), order);
            } catch (Exception e) {
                log.error("Email notification failed for order {}", order.getId(), e);
            } finally {
                MDC.clear();
            }
        }, notificationExecutor);
 
        CompletableFuture.runAsync(() -> {
            MDC.put("correlationId", correlationId);
            try {
                smsService.sendOrderSms(user.getPhoneNumber(), order);
            } catch (Exception e) {
                log.warn("SMS notification failed for order {}", order.getId(), e);
            } finally {
                MDC.clear();
            }
        }, notificationExecutor);
 
        CompletableFuture.runAsync(() -> {
            MDC.put("correlationId", correlationId);
            try {
                pushService.sendPushNotification(user.getDeviceToken(), order);
            } catch (Exception e) {
                log.warn("Push notification failed for order {}", order.getId(), e);
            } finally {
                MDC.clear();
            }
        }, notificationExecutor);
 
        log.info("Notification dispatch initiated for order {}", order.getId());
        // Returns immediately. Notifications are sent in background.
    }
}

@Service
@Slf4j
public class BatchOrderProcessor {
 
    /**
     * Process 10,000 orders in parallel with controlled concurrency.
     * Too much parallelism overwhelms the database.
     * Use partitioning to process in batches of manageable size.
     */
    public BatchResult processOrders(List<Long> orderIds) {
        int batchSize = 100;  // Process 100 orders at a time
        List<List<Long>> batches = Lists.partition(orderIds, batchSize);
 
        List<BatchPartialResult> allResults = new ArrayList<>();
        AtomicInteger processedCount = new AtomicInteger(0);
 
        for (List<Long> batch : batches) {
            // Process each order in the current batch in parallel
            List<CompletableFuture<OrderProcessingResult>> batchFutures = batch.stream()
                .map(orderId ->
                    CompletableFuture.supplyAsync(
                        () -> processSingleOrder(orderId),
                        ioExecutor
                    )
                    .orTimeout(30, TimeUnit.SECONDS)
                    .exceptionally(ex -> {
                        log.error("Failed to process order {}: {}", orderId, ex.getMessage());
                        return OrderProcessingResult.failed(orderId, ex.getMessage());
                    })
                )
                .toList();
 
            // Wait for the entire batch before starting the next batch
            // This prevents overwhelming the database with 10,000 simultaneous queries
            List<OrderProcessingResult> batchResults = CompletableFuture
                .allOf(batchFutures.toArray(new CompletableFuture[0]))
                .thenApply(v -> batchFutures.stream()
                    .map(f -> f.getNow(null))
                    .filter(Objects::nonNull)
                    .toList()
                )
                .join();
 
            int processed = processedCount.addAndGet(batchResults.size());
            log.info("Processed {}/{} orders", processed, orderIds.size());
 
            allResults.add(aggregateBatchResults(batchResults));
        }
 
        return aggregateAllResults(allResults);
    }
}

@Service
@Slf4j
public class InventoryService {
 
    private final CircuitBreaker inventoryApiCircuitBreaker;
    private final ExecutorService httpExecutor;
 
    /**
     * Combine CompletableFuture with Resilience4j circuit breaker.
     * If the inventory API is down, circuit opens and fallback is used immediately.
     */
    public CompletableFuture<InventoryStatus> checkInventory(Long productId) {
        return CompletableFuture.supplyAsync(
            () -> inventoryApiCircuitBreaker.executeSupplier(
                () -> inventoryApiClient.getStatus(productId)
            ),
            httpExecutor
        )
        .orTimeout(2, TimeUnit.SECONDS)
        .exceptionally(ex -> {
            if (ex.getCause() instanceof CallNotPermittedException) {
                log.warn("Circuit OPEN for inventory API. Product {}: using cached status.", productId);
                return inventoryCache.getLastKnownStatus(productId);
            }
            log.error("Inventory check failed for product {}: {}", productId, ex.getMessage());
            return InventoryStatus.unknown(productId);
        });
    }
}

@Service
@Slf4j
public class ExternalApiService {
 
    private final WebClient webClient;
 
    /**
     * Call external API using WebClient and get a CompletableFuture back.
     * This is the bridge between reactive (WebClient) and imperative (CompletableFuture).
     */
    public CompletableFuture<ExternalData> fetchExternalData(String resourceId) {
        return webClient.get()
            .uri("/api/data/{id}", resourceId)
            .header("X-Correlation-Id", RequestContext.getCorrelationId())
            .retrieve()
            .onStatus(
                status -> status.is4xxClientError(),
                response -> response.bodyToMono(ErrorResponse.class)
                    .map(err -> new ResourceNotFoundException(resourceId))
            )
            .onStatus(
                status -> status.is5xxServerError(),
                response -> Mono.error(new ExternalServiceException("External API server error"))
            )
            .bodyToMono(ExternalData.class)
            .timeout(Duration.ofSeconds(3))
            .toFuture();  // Convert Mono<T> -> CompletableFuture<T>
    }
 
    /**
     * Multiple parallel external API calls using WebClient + allOf.
     */
    public CompletableFuture<AggregatedData> fetchMultipleResources(List<String> ids) {
        List<CompletableFuture<ExternalData>> futures = ids.stream()
            .map(this::fetchExternalData)
            .toList();
 
        return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> {
                List<ExternalData> results = futures.stream()
                    .map(CompletableFuture::join)
                    .toList();
                return new AggregatedData(results);
            });
    }
}

// Pattern 1: Snapshot and restore manually (verbose but explicit)
public CompletableFuture<Order> placeOrderAsync(OrderRequest request) {
    Map<String, String> mdcSnapshot = MDC.getCopyOfContextMap();  // Capture now
 
    return CompletableFuture.supplyAsync(() -> {
        MDC.setContextMap(mdcSnapshot != null ? mdcSnapshot : Map.of());  // Restore
        try {
            return orderService.create(request);
        } finally {
            MDC.clear();  // Clean up worker thread
        }
    }, ioExecutor);
}
 
// Pattern 2: ContextAwareExecutor wrapper (cleaner - see threadlocal guide)
return CompletableFuture.supplyAsync(
    () -> orderService.create(request),
    contextAwareExecutor  // Handles snapshot/restore automatically
);
 
// Pattern 3: TaskDecorator on Spring @Async executor (for @Async methods)
// - Configured in AsyncConfig (see threadlocal guide)
 
// Pattern 4: TransmittableThreadLocal + TtlExecutors (most automatic)
// - Replace ThreadLocal with TransmittableThreadLocal
// - Wrap executor with TtlExecutors.getTtlExecutorService(pool)
// - Context propagates automatically without any code changes

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new IllegalArgumentException("Bad argument");
});
 
// WRONG: catching the original exception type
try {
    future.join();
} catch (IllegalArgumentException e) {
    // This NEVER catches! join() wraps in CompletionException.
    System.out.println("Caught: " + e.getMessage());
}
 
// CORRECT: catch CompletionException and unwrap
try {
    future.join();
} catch (CompletionException e) {
    Throwable cause = e.getCause();  // The actual exception
    if (cause instanceof IllegalArgumentException) {
        System.out.println("Caught: " + cause.getMessage());
    }
}
 
// CLEANER: use .get() which throws the actual exception (as ExecutionException)
try {
    future.get();
} catch (ExecutionException e) {
    Throwable cause = e.getCause();  // The actual exception
    if (cause instanceof IllegalArgumentException) {
        System.out.println("Caught: " + cause.getMessage());
    }
} catch (InterruptedException e) {
    Thread.currentThread().interrupt();
}

CompletableFuture<String> result =
    CompletableFuture.supplyAsync(() -> {
        throw new RuntimeException("Stage 1 failed");
    })
    .thenApply(x -> {
        System.out.println("Stage 2 - SKIPPED because stage 1 failed");
        return x + " transformed";
    })
    .thenApply(x -> {
        System.out.println("Stage 3 - ALSO SKIPPED");
        return x;
    })
    .exceptionally(ex -> {
        System.out.println("Caught in exceptionally: " + ex.getMessage());
        return "fallback value";  // Chain continues from here with "fallback value"
    })
    .thenApply(x -> {
        System.out.println("Stage 4 runs: " + x);  // Runs with "fallback value"
        return x.toUpperCase();
    });
 
// Output:
// Caught in exceptionally: java.lang.RuntimeException: Stage 1 failed
// Stage 4 runs: fallback value

@SpringBootTest
class ProductPageServiceTest {
 
    @Autowired
    private ProductPageService productPageService;
 
    @MockBean
    private ReviewServiceClient reviewClient;
 
    @MockBean
    private InventoryClient inventoryClient;
 
    /**
     * Test 1: Normal case - all futures complete successfully.
     */
    @Test
    void getProductPage_allServicesAvailable_returnsFullPage() {
        // Arrange
        when(reviewClient.getSummary(1L)).thenReturn(new ReviewSummary(4.5, 100));
        when(inventoryClient.getInfo(1L)).thenReturn(new InventoryInfo(50, true));
 
        // Act
        ProductPage page = productPageService.getProductPage(1L, 42L);
 
        // Assert
        assertThat(page.getReviews().getAverageRating()).isEqualTo(4.5);
        assertThat(page.getInventory().isAvailable()).isTrue();
    }
 
    /**
     * Test 2: Review service down - should return page with empty reviews (graceful degradation).
     */
    @Test
    void getProductPage_reviewServiceDown_returnPageWithEmptyReviews() {
        // Arrange
        when(reviewClient.getSummary(any())).thenThrow(new RuntimeException("Service unavailable"));
        when(inventoryClient.getInfo(any())).thenReturn(new InventoryInfo(10, true));
 
        // Act - should not throw
        ProductPage page = productPageService.getProductPage(1L, 42L);
 
        // Assert - page loaded with empty reviews, not an exception
        assertThat(page.getReviews()).isEqualTo(ReviewSummary.empty());
        assertThat(page.getInventory().isAvailable()).isTrue();  // Other data intact
    }
 
    /**
     * Test 3: Timeout simulation using a slow mock.
     */
    @Test
    void getProductPage_reviewServiceTimesOut_returnsPageWithEmptyReviews()
            throws InterruptedException {
 
        // Arrange: review service takes 1 second (longer than 300ms timeout)
        when(reviewClient.getSummary(any())).thenAnswer(invocation -> {
            Thread.sleep(1000);  // Simulate slow response
            return new ReviewSummary(4.0, 50);
        });
        when(inventoryClient.getInfo(any())).thenReturn(new InventoryInfo(5, true));
 
        // Act
        long start = System.currentTimeMillis();
        ProductPage page = productPageService.getProductPage(1L, 42L);
        long elapsed = System.currentTimeMillis() - start;
 
        // Assert: page returned quickly (not waiting for 1s timeout)
        assertThat(elapsed).isLessThan(600L);  // Within 600ms (not 1000ms)
        assertThat(page.getReviews()).isEqualTo(ReviewSummary.empty());
    }
 
    /**
     * Test 4: CompletableFuture that we create manually.
     */
    @Test
    void completedFuture_immediatelyAvailable() throws ExecutionException, InterruptedException {
        // Test using already-completed futures (no async needed)
        CompletableFuture<String> future = CompletableFuture.completedFuture("test-value");
 
        assertThat(future.isDone()).isTrue();
        assertThat(future.get()).isEqualTo("test-value");
    }
 
    /**
     * Test 5: Exception propagation.
     */
    @Test
    void failedFuture_propagatesException() {
        CompletableFuture<String> failed =
            CompletableFuture.failedFuture(new IllegalArgumentException("Bad input"));
 
        assertThatThrownBy(() -> failed.join())
            .isInstanceOf(CompletionException.class)
            .hasCauseInstanceOf(IllegalArgumentException.class)
            .hasMessageContaining("Bad input");
    }
}

// ANTI-PATTERN: Start async, then immediately block. Zero benefit over synchronous.
String result = CompletableFuture.supplyAsync(() -> callService())
    .get();  // Blocks immediately. You gained nothing by going async.
 
// CORRECT: Chain callbacks. Only block at the outermost level if absolutely needed.
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> callService())
    .thenApply(this::transform)
    .thenApply(this::format);
// ... later, if you must block:
String result = future.join();

// ANTI-PATTERN: Launch 10,000 tasks simultaneously
List<CompletableFuture<Result>> futures = orderIds.stream()
    .map(id -> CompletableFuture.supplyAsync(() -> processOrder(id), ioExecutor))
    .toList();
// Queues 10,000 tasks. If each holds a DB connection, you exhaust the connection pool.
 
// CORRECT: Process in controlled batches (see Batch Processing example above)
List<List<Long>> batches = Lists.partition(orderIds, 100);
for (List<Long> batch : batches) {
    processBatch(batch);  // Wait for each batch before starting next
}

// ANTI-PATTERN: Using default executor (ForkJoinPool) for database I/O
CompletableFuture.supplyAsync(() -> userRepository.findById(id));
// ForkJoinPool has CPU_cores - 1 threads. With 4 CPUs: 3 threads.
// 3 slow DB queries can stall your entire application.
 
// CORRECT: Always use a dedicated I/O executor
CompletableFuture.supplyAsync(() -> userRepository.findById(id), ioExecutor);

// ANTI-PATTERN: Exceptions are swallowed silently
CompletableFuture.runAsync(() -> sendEmail(user));  // If this throws, you never know
 
// CORRECT: Add exception handling even for fire-and-forget
CompletableFuture.runAsync(() -> sendEmail(user))
    .exceptionally(ex -> {
        log.error("Failed to send email to {}: {}", user.getEmail(), ex.getMessage());
        alertService.notify("EMAIL_FAILURE", ex);
        return null;
    });

// Sequential: 100ms + 150ms + 80ms = 330ms total
User user         = userService.getUser(userId);           // 100ms
List<Order> orders = orderService.getOrders(userId);        // 150ms
List<Product> recs  = recommendationService.get(userId);    // 80ms
 
// Parallel with allOf: max(100ms, 150ms, 80ms) = 150ms total (2.2x faster)
CompletableFuture<User> userFuture     = supplyAsync(() -> userService.getUser(userId),     ioExec);
CompletableFuture<List<Order>> ordFuture = supplyAsync(() -> orderService.getOrders(userId), ioExec);
CompletableFuture<List<Product>> recFuture = supplyAsync(() -> recService.get(userId),      ioExec);
 
CompletableFuture.allOf(userFuture, ordFuture, recFuture).join();
 
User user          = userFuture.join();
List<Order> orders = ordFuture.join();
List<Product> recs = recFuture.join();

// Create
CompletableFuture.supplyAsync(supplier, executor)   // returns a value
CompletableFuture.runAsync(runnable, executor)       // returns void
CompletableFuture.completedFuture(value)             // already done
CompletableFuture.failedFuture(exception)            // already failed
 
// Transform (no new thread)
.thenApply(fn)       // T -> U (sync transform)
.thenAccept(fn)      // T -> void (consume)
.thenRun(fn)         // () -> void (ignore result)
 
// Transform (new thread from pool)
.thenApplyAsync(fn, executor)
.thenAcceptAsync(fn, executor)
.thenRunAsync(fn, executor)
 
// Compose (chain async operations)
.thenCompose(fn)     // T -> CF<U>  (flatMap - avoids nesting)
.thenCombine(other, fn)  // (T, U) -> V  (merge two parallel futures)
 
// Combine many
CompletableFuture.allOf(f1, f2, f3)    // wait for ALL
CompletableFuture.anyOf(f1, f2, f3)    // wait for FIRST
 
// Error handling
.exceptionally(fn)   // only on failure -> recovery value
.handle(fn)          // always -> (result or null, exception or null)
.whenComplete(fn)    // always -> side effect, does not change result
 
// Timeout (Java 9+)
.orTimeout(n, unit)                 // throw TimeoutException if late
.completeOnTimeout(value, n, unit)  // return default value if late
 
// Get the result
.join()              // block, throws CompletionException (unchecked)
.get()               // block, throws ExecutionException (checked)
.getNow(default)     // return now or default if not done
.isDone()            // check completion without blocking