Introduction
Java's CompletableFuture, introduced in Java 8, revolutionized asynchronous programming by providing a powerful framework for composing asynchronous operations. However, one common challenge developers face is implementing proper timeout handling to prevent operations from hanging indefinitely. This comprehensive guide explores various techniques for adding timeout capabilities to CompletableFuture operations.
Why Timeout Handling is Crucial
In distributed systems and microservices architectures, network calls and external service dependencies can fail silently or hang. Without proper timeout mechanisms:
- Resources may be tied up indefinitely
- Application performance can degrade
- Cascading failures may occur
- User experience suffers from unresponsive behavior
Basic Timeout with orTimeout()
Java 9 introduced the orTimeout() method, which provides a straightforward way to add timeout behavior to CompletableFuture.
Syntax
public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit)
Example
import java.util.concurrent.*;
public class BasicTimeoutExample {
public static void main(String[] args) {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
// Simulate long-running task
Thread.sleep(5000);
return "Task completed successfully";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
})
.orTimeout(2, TimeUnit.SECONDS) // Timeout after 2 seconds
.exceptionally(throwable -> {
if (throwable instanceof TimeoutException) {
return "Task timed out";
}
return "Other error: " + throwable.getMessage();
});
System.out.println(future.join()); // Output: "Task timed out"
}
}
Complete Timeout with completeOnTimeout()
The completeOnTimeout() method allows you to provide a default value when a timeout occurs, rather than throwing an exception.
Syntax
public CompletableFuture<T> completeOnTimeout(T value, long timeout, TimeUnit unit)
Example
public class CompleteOnTimeoutExample {
public static void main(String[] args) {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
return "Actual result";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
})
.completeOnTimeout("Default value", 2, TimeUnit.SECONDS);
System.out.println(future.join()); // Output: "Default value"
}
}
Advanced Timeout Patterns
1. Custom Executor with Timeout
public class CustomExecutorTimeout {
private static final ScheduledExecutorService scheduler =
Executors.newScheduledThreadPool(1);
public static <T> CompletableFuture<T> withTimeout(
CompletableFuture<T> future, long timeout, TimeUnit unit) {
return future.whenComplete((result, throwable) -> {
if (!future.isDone()) {
// Cancel the original future if not completed
future.cancel(true);
}
}).orTimeout(timeout, unit);
}
public static void main(String[] args) {
CompletableFuture<String> longRunningTask = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(5000);
return "Long running result";
} catch (InterruptedException e) {
throw new RuntimeException("Task interrupted", e);
}
});
CompletableFuture<String> timeoutFuture = withTimeout(longRunningTask, 2, TimeUnit.SECONDS);
try {
String result = timeoutFuture.get();
System.out.println("Result: " + result);
} catch (Exception e) {
System.out.println("Exception: " + e.getCause());
}
}
}
2. Multiple Operations with Individual Timeouts
public class MultipleTimeouts {
public static void main(String[] args) {
// Simulate multiple service calls with different timeouts
CompletableFuture<String> service1 = callService("Service1", 3000, 2000);
CompletableFuture<String> service2 = callService("Service2", 1000, 3000);
CompletableFuture<String> service3 = callService("Service3", 4000, 1500);
// Combine results
CompletableFuture<Void> allResults = CompletableFuture.allOf(service1, service2, service3);
// Handle individual results
service1.whenComplete((result, error) ->
handleResult("Service1", result, error));
service2.whenComplete((result, error) ->
handleResult("Service2", result, error));
service3.whenComplete((result, error) ->
handleResult("Service3", result, error));
allResults.join();
}
private static CompletableFuture<String> callService(String serviceName,
long processingTime,
long timeout) {
return CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(processingTime);
return serviceName + " result";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).orTimeout(timeout, TimeUnit.MILLISECONDS)
.exceptionally(throwable -> serviceName + " failed: " + throwable.getMessage());
}
private static void handleResult(String serviceName, String result, Throwable error) {
if (error != null) {
System.out.println(serviceName + " error: " + error.getMessage());
} else {
System.out.println(serviceName + " success: " + result);
}
}
}
Timeout in Composition Chains
Handling Timeouts in Complex Pipelines
public class CompositionTimeout {
public static void main(String[] args) {
CompletableFuture<String> pipeline = CompletableFuture
.supplyAsync(() -> {
try {
Thread.sleep(1000);
return "Initial data";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
})
.orTimeout(500, TimeUnit.MILLISECONDS) // Timeout for initial operation
.thenApplyAsync(data -> {
try {
Thread.sleep(800);
return data + " -> processed";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
})
.orTimeout(1000, TimeUnit.MILLISECONDS) // Timeout for processing step
.exceptionally(throwable -> {
if (throwable instanceof TimeoutException) {
return "Fallback value due to timeout";
}
return "Fallback due to other error: " + throwable.getMessage();
});
System.out.println("Final result: " + pipeline.join());
}
}
Best Practices for Timeout Handling
1. Choose Appropriate Timeout Values
public class TimeoutConfig {
// Different timeouts for different types of operations
public static final long DATABASE_TIMEOUT_MS = 5000;
public static final long EXTERNAL_SERVICE_TIMEOUT_MS = 10000;
public static final long CACHE_TIMEOUT_MS = 100;
public static CompletableFuture<String> callDatabase() {
return CompletableFuture.supplyAsync(DatabaseService::query)
.orTimeout(DATABASE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
}
public static CompletableFuture<String> callExternalService() {
return CompletableFuture.supplyAsync(ExternalService::call)
.orTimeout(EXTERNAL_SERVICE_TIMEOUT_MS, TimeUnit.MILLISECONDS);
}
}
2. Graceful Fallback Strategies
public class FallbackStrategies {
public static CompletableFuture<String> robustServiceCall() {
return CompletableFuture.supplyAsync(() -> {
// Primary service call
try {
Thread.sleep(2000);
return "Primary service result";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
})
.orTimeout(1000, TimeUnit.MILLISECONDS)
.exceptionally(throwable -> {
// Fallback to secondary service
System.out.println("Primary service failed, trying secondary...");
return callSecondaryService().join();
})
.exceptionally(throwable -> {
// Final fallback to cached value
System.out.println("All services failed, using cache");
return getCachedValue();
});
}
private static CompletableFuture<String> callSecondaryService() {
return CompletableFuture.supplyAsync(() -> "Secondary service result")
.orTimeout(500, TimeUnit.MILLISECONDS);
}
private static String getCachedValue() {
return "Cached value";
}
}
3. Resource Cleanup
public class ResourceCleanup {
public static CompletableFuture<String> withResourceCleanup() {
AtomicReference<ExternalResource> resource = new AtomicReference<>();
return CompletableFuture.supplyAsync(() -> {
resource.set(acquireResource());
try {
// Simulate work
Thread.sleep(3000);
return "Work completed";
} finally {
// Ensure resource cleanup
if (resource.get() != null) {
resource.get().cleanup();
}
}
})
.orTimeout(2000, TimeUnit.MILLISECONDS)
.whenComplete((result, error) -> {
// Additional cleanup in case of timeout
if (error instanceof TimeoutException && resource.get() != null) {
resource.get().emergencyCleanup();
}
});
}
static class ExternalResource {
void cleanup() { System.out.println("Normal cleanup"); }
void emergencyCleanup() { System.out.println("Emergency cleanup"); }
}
static ExternalResource acquireResource() {
return new ExternalResource();
}
}
Testing Timeout Behavior
Unit Testing with Mocks
public class TimeoutTest {
@Test
public void testTimeoutBehavior() {
// Given a task that takes longer than timeout
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
return "Success";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).orTimeout(500, TimeUnit.MILLISECONDS);
// When & Then - expect timeout exception
assertThrows(CompletionException.class, () -> {
future.join();
});
}
@Test
public void testCompleteOnTimeout() {
// Given a task with completeOnTimeout
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(2000);
return "Actual";
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}).completeOnTimeout("Default", 500, TimeUnit.MILLISECONDS);
// Then - should return default value
assertEquals("Default", future.join());
}
}
Performance Considerations
1. Thread Pool Management
public class OptimizedTimeoutHandling {
private static final ScheduledExecutorService timeoutExecutor =
Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors());
private static final ExecutorService workerExecutor =
Executors.newFixedThreadPool(10);
public static <T> CompletableFuture<T> optimizedWithTimeout(
Callable<T> task, long timeout, TimeUnit unit) {
CompletableFuture<T> future = CompletableFuture.supplyAsync(() -> {
try {
return task.call();
} catch (Exception e) {
throw new RuntimeException(e);
}
}, workerExecutor);
return future.orTimeout(timeout, unit);
}
}
2. Monitoring and Metrics
public class MonitoredTimeouts {
private static final Counter timeoutCounter =
Metrics.counter("completablefuture.timeouts");
public static <T> CompletableFuture<T> monitoredWithTimeout(
CompletableFuture<T> future, String operation, long timeout, TimeUnit unit) {
return future.orTimeout(timeout, unit)
.whenComplete((result, throwable) -> {
if (throwable instanceof TimeoutException) {
timeoutCounter.increment();
System.out.println("Timeout occurred for operation: " + operation);
}
});
}
}
Common Pitfalls and How to Avoid Them
1. Forgetting to Handle Timeout Exceptions
Bad Practice:
CompletableFuture<String> future = someAsyncCall() .orTimeout(1, TimeUnit.SECONDS); String result = future.get(); // Throws exception on timeout
Good Practice:
CompletableFuture<String> future = someAsyncCall() .orTimeout(1, TimeUnit.SECONDS) .exceptionally(throwable -> "Fallback value");
2. Incorrect Timeout Placement
Bad Practice:
CompletableFuture<String> future = someAsyncCall() .thenApply(result -> transform(result)) .orTimeout(1, TimeUnit.SECONDS); // Only times out the transformation
Good Practice:
CompletableFuture<String> future = someAsyncCall() .orTimeout(1, TimeUnit.SECONDS) // Timeout includes the async call .thenApply(result -> transform(result));
Conclusion
Effective timeout handling in CompletableFuture is essential for building robust, responsive applications. By leveraging methods like orTimeout() and completeOnTimeout(), along with proper fallback strategies and resource management, you can create systems that gracefully handle delays and failures.
Remember to:
- Choose timeout values appropriate for each operation type
- Implement comprehensive fallback mechanisms
- Always clean up resources properly
- Monitor timeout occurrences for system health
- Test timeout behavior thoroughly
With these techniques, you'll be well-equipped to handle the challenges of asynchronous programming in Java while maintaining system reliability and performance.