Improving microservice resilience in Liberty

Circuit breaker code snippet 1: Create a CircuitBreakerBean with configured CircuitBreaker and Timeout.

@RequestScoped
public class CircuitBreakerBean {

    private int executionCounterA = 0;

    // The combined effect of the specified requestVolumeThreshold and failureRatio is that 3  
    // failures will trigger the circuit to open.
    // After a 1 second delay the Circuit will allow fresh attempts to invoke the service.
    @CircuitBreaker(delay = 1, delayUnit = ChronoUnit.SECONDS, requestVolumeThreshold = 3, failureRatio = 1.0)
    // A service is considered to have timed out after 3 seconds
    @Timeout(value = 3, unit = ChronoUnit.SECONDS)
    public String serviceA() {
        executionCounterA++;

        if (executionCounterA <= 3) {
            //Sleep for 10 secs to force a timeout
            try {
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                System.out.println("serviceA interrupted");
            }
        }

CircuitBreaker code snippet 2: Using the CircuitBreakerBean.

@Inject
CircuitBreakerBean bean;

// FaultTolerance bean with circuit breaker, should fail 3 times
for (int i = 0; i < 3; i++) {
    try {
        bean.serviceA();
        throw new AssertionError("TimeoutException not caught");
    } catch (TimeoutException e) {
            //expected
    }
}

// The CircuitBreaker should be open, so calling serviceA should generate a 
// CircuitBreakerOpenException.
try {
    bean.serviceA();
    throw new AssertionError("CircuitBreakerOpenException not caught");
} catch (CircuitBreakerOpenException e) {
    //expected
}

//allow time for the circuit to re-close
Thread.sleep(3000);

// The CircuitBreaker should be closed and serviceA should now succeed.
String res = bean.serviceA();
if (!"serviceA: 4".equals(res)) {
    throw new AssertionError("Bad Result: " + res);
}

Fallback and Retry code snippet 1: FTServiceBean with configured FallbackHandler and Retry policy.

@RequestScoped
public class FTServiceBean {

    // Annotate serviceA with a named FallbackHandler and a Retry policy specifying the
    // number of retries.
    @Retry(maxRetries = 2)
    @Fallback(StringFallbackHandler.class)
    public String serviceA() {
        throw new RuntimeException("Connection failed");
        return null;
    }   
}

Fallback and Retry code snippet 2: A FallbackHandler, the code that is driven if the main service fails.

@Dependent
public class StringFallbackHandler implements FallbackHandler<String> {

    @Override
    public String handle(ExecutionContext context) {
        return "fallback for " + context.getMethod().getName();
    }
}

Fallback and Retry code snippet 3: Using the FTServiceBean.

private @Inject FTServiceBean ftServiceBean;
    
try {
    // Call serviceA, which will be retried twice in the event of failure, after which
    // the FallbackHandler will be driven.
    String result = ftServiceBean.serviceA();
    if(!result.contains("serviceA"))
       throw new AssertionError("The message should be \"fallback for serviceA\"");
 }
catch(RuntimeException ex) {
    throw new AssertionError("serviceA should not throw a RuntimeException");
}

@RequestScoped
public class AsynchronousBean {
    
    @Asynchronous
    public Future<String> serviceA() {
        try {
            // Sleep to simulate work
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            throw new RuntimeException("serviceA interrupted", e);
        }
        // Return the result in a completed CompletableFuture
        return CompletableFuture.completedFuture("serviceA OK");
    }
    
    // Note: returning a CompletionStage requires Fault Tolerance 2.0
    @Asynchronous
    public CompletionStage<String> serviceB() {
        try {
            // Sleep to simulate work
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            throw new RuntimeException("serviceB interrupted", e);
        }
        // Return the result in a completed CompletableFuture (which implements CompletionStage)
        return CompletableFuture.completedFuture("serviceB OK");
    }

}

@Inject AsynchronousBean asyncBean;

// serviceA and serviceB methods will run in parallel because they are annotated with @Asynchronous
Future<String> resultA = asyncBean.serviceA();
CompletionStage<String> resultB = asyncBean.serviceB();

// The CompletionStage returned from serviceB allows us to add actions which take place when the serviceB method finishes
resultB.thenAccept((r) -> System.out.println("ServiceB result: " + r))
       .exceptionally((ex) -> {
           System.out.println("ServiceB failed");
           ex.printStackTrace();
           return null;
       });

// For the Future returned from serviceA, we need to wait for it to finish, then we can handle the result
try {
    System.out.println("serviceA result: " + resultA.get());
} catch (ExecutionException ex) {
    System.out.println("ServiceA failed");
    ex.printStackTrace();
} catch (InterruptedException ex) {
    System.out.println("Interrupted waiting for serviceA");
}

Bulkhead code snippet 1: Create a BulkheadBean with configured Bulkhead.

@RequestScoped
@Asynchronous
public class BulkheadBean {

    private final AtomicInteger connectATokens = new AtomicInteger(0);

    // Configure a Bulkhead that supports at most 2 concurrent threads.
    @Bulkhead(maxThreads = 2)
    public Future<Boolean> connectA(String data) throws InterruptedException {
        System.out.println("connectA starting " + data);
        int token = connectATokens.incrementAndGet();
        try {
            if (token > 2) {
                throw new RuntimeException("Too many threads in connectA[" + data + "]: " + token);
            }
            Thread.sleep(5000);
            return CompletableFuture.completedFuture(Boolean.TRUE);
        } finally {
            connectATokens.decrementAndGet();
            System.out.println("connectA complete " + data);
        }
    }
}

Bulkhead code snippet 2: Use the BulkheadBean.

@Inject
BulkheadBean bean;

// connectA has a poolSize of 2
// The first two calls to connectA should be run straight away, in parallel, each around
// 5 seconds
Future<Boolean> future1 = bean.connectA("One");
Thread.sleep(100);
Future<Boolean> future2 = bean.connectA("Two");
Thread.sleep(100);

// The next two calls to connectA should wait until the first 2 have finished
Future<Boolean> future3 = bean.connectA("Three");
Thread.sleep(100);
Future<Boolean> future4 = bean.connectA("Four");
Thread.sleep(100);

//total time should be just over 10s
Thread.sleep(11000);

if (!future1.get(1000, TimeUnit.MILLISECONDS)) {
    throw new AssertionError("Future1 did not complete properly");
}
if (!future2.get(1000, TimeUnit.MILLISECONDS)) {
    throw new AssertionError("Future2 did not complete properly");
}
if (!future3.get(1000, TimeUnit.MILLISECONDS)) {
    throw new AssertionError("Future3 did not complete properly");
}
if (!future4.get(1000, TimeUnit.MILLISECONDS)) {
    throw new AssertionError("Future4 did not complete properly");
}