resilience4j: Best implementation for circuitBreaker + ThreadPoolBulkhead + timeLimiter + retry

Hi! I’m migrating a task from hystrix to resilience4j where I need the circuitBreaker, ThreadPoolBulkhead and timeLimiter with a fallback. I was able to have a good implementation for circuitBreaker and timelimiter, but now, that I’m adding the needed ThreadPoolBulkhead, I’m not sure if the implementation is the best:

    protected Future<ErrorCodes> execute() throws ExecutionException, InterruptedException {
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("sample", "circuitBreakerConfig");
        ThreadPoolBulkhead bulkhead = bulkheadRegistry.bulkhead("sample", "threadPoolBulkheadConfigForTransactions");

        Supplier<CompletionStage<ErrorCodes>> completionStageSupplier = () -> CompletableFuture.supplyAsync(this::run);
        Supplier<CompletionStage<ErrorCodes>> decoratedCompletionStage = Decorators.ofCompletionStage(completionStageSupplier)
                .withCircuitBreaker(circuitBreaker)
                .decorate();

        CompletableFuture<ErrorCodes> bulkheadExecutor = bulkhead.executeSupplier(decoratedCompletionStage).toCompletableFuture().get().toCompletableFuture();
        Callable<ErrorCodes> result = TimeLimiter.decorateFutureSupplier(TimeLimiter.of(timeLimiterConfig), () -> bulkheadExecutor);

        return Future.of(result::call).recover(this::getFallback);
    }

About this issue

Original URL
State: closed
Created 4 years ago
Comments: 23 (12 by maintainers)

Most upvoted comments

Can I close the issue?

Sure! Thanks!

IsabelPalomar on Jan 31, 2020

Even better:

CompletableFuture<String> future = Decorators
    .ofSupplier(() -> helloWorldService.returnHelloWorld())
    .withThreadPoolBulkhead(ThreadPoolBulkhead.ofDefaults("helloBackend"))
    .withTimeLimiter(TimeLimiter.ofDefaults(), Executors.newSingleThreadScheduledExecutor())
    .withCircuitBreaker(circuitBreaker)
    .withFallback(TimeoutException.class, (e) -> "Recovery")
    .get().toCompletableFuture();

RobWin on Jan 31, 2020

Thank you so much for all the examples! I was able to run my tasks and the tests for the different configs. It really helped me to put the completionStage.exceptionally(exceptionHandler), the code looks cleaner. Looking forward the next release! 🙌

IsabelPalomar on Jan 31, 2020

The next release version will make it simpler. I copied some of the code into a new CompletionStageUtils class and enhanced the Decorators builder by a withTimeLimiter method.

private <T> CompletableFuture<T> executeAsync(Supplier<T> supplier, Function<Throwable, T> exceptionHandler,
        ScheduledExecutorService scheduler){
     ThreadPoolBulkhead bulkhead = bulkheadRegistry.bulkhead("sample");
     TimeLimiter timeLimiter = timeLimiterRegistry.timeLimiter("sample);
     CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("sample");
     Retry retry = retryRegistry.retry("sample");

    CompletionStage<T> completionStage = Decorators
            .ofCompletionStage(threadPoolBulkhead.decorateSupplier(supplier))
            .withTimeLimiter(timeLimiter, scheduler)
            .withCircuitBreaker(circuitBreaker)
            .withRetry(retry)
            .get();

    return CompletionStageUtils.recover(completionStage, exceptionHandler)
            .toCompletableFuture();
}

RobWin on Jan 30, 2020