DEV Community: Gabriel Aramburu

Spring Cloud Loadbalancer with Eureka Server

Gabriel Aramburu — Fri, 23 Sep 2022 20:43:37 +0000

Introduction

I built a very simple application that takes advantage of the Eureka Discovery Service.

It consists of a web application that needs to consume an external service in order to complete his work.

At some point, there will be several service instances running on different machines. The web application uses Eureka to discover the available service instances.

The idea is to simulate an application that reacts to a high load situation, adding more services in order to maintain the response time under certain limits.

I found some difficulties trying to build a version that works well under relatively high load so I will describe the solution that worked for me.

I use Spring Cloud 2021.0.4 and Spring Boot 2.6.11 version.

The problem

When several service instances were running and processing the requests sent by the web application, If one of them was shut down the loadbalancer still sent requests to him.

Therefore the application was not able to process the request and an error was sent to the client.

This graph belongs to JMeter, which was used to simulate concurrent application’s clients.

The green line represents the number of failed transactions that the clients received after one of the service instances was shut down.

After several seconds the loadbalancer started to send requests to available service instances again and the problem was solved.

Expected behavior

The loadbalancer executing in the web application should only send requests to available service instances.The client should not receive an error when a service instance is shut down.

First intent.

The time period where the failed requests were registered corresponds to the loadbalancer cache time to live (ttl) configuration. So decreasing the cache refresh time improves the situation.

#Configuration of web client application.
spring.cloud.loadbalancer.cache.ttl=10s

To decrease the loadbalancer cache expiration time seems to be a work around due to the fact that the loadbalancer should avoid sending requests to an unavailable instance even with an out of date cache.

Second intent.

Another possible idea was to surround the RestTemplate call with a retry mechanism using a loop or an already retry implementation like Resilience4j, which is already integrated with Spring Boot.

The problem with this approach is that the loadbalancer is a singleton component and shares the state with all the incoming requests.

When the balancer uses a Round Robin algorithm (even the problem applies to other implementations) there is no guarantee that when a request is sent to an unavailable instance the second try will not be processed for the same instance again and again.

Actually with few service instances it is very easy to reproduce the problem.

Third intent

According to the Spring Cloud documentation we need to add a dependency to our web client application.

        <dependency>
            <groupId>org.springframework.retry</groupId>
            <artifactId>spring-retry</artifactId>
        </dependency>

When present, this retry implementation is used internally by the Spring Cloud loadbalancer component.

Now, after running the first test again, the client does not get any error when a service instance is shut down. As the graph shows, the web application throughput was affected for a moment, but at least the client did not receive any failed request.

As the metrics show, the loadbalancer still is sending request to the unavailable instance, but at least in conjunction with Spring Retry is able to deal with the ConnectionException in a more accurate way.

Finally, the solution that worked better for me was to disable the LoadBalancer cache an to increase the retries max to three.

spring.cloud.loadbalancer.cache.enabled=false
spring.cloud.loadbalancer.retry.max-retries-on-next-service-instance=3

Conclusion

It is mandatory to include the Spring Retry dependency in order to improve the resilience of the Spring Cloud loadbalancer implementation.

Bulkhead pattern: semaphore vs threadPool

Gabriel Aramburu — Mon, 29 Aug 2022 17:16:00 +0000

Why two different implementations?

In this article I will refer to Spring and Resilence4j.

If you are reading this article you probably already know about the bulkhead pattern, what is the problem that it intends to solve and the most common implementations: semaphore based and threadPool based.

At least for me, it was not easy to realize when to use a semaphore implementation and when to use a threadPool one.

I know how a Semaphore works, and I also understand the ThreadPool pattern, therefore a short answer and the most obvious could have been: use a threadPool for limiting the number of asynchronous calls and use a semaphore for limiting synchronous ones.

So why was the difficult part? The reason was these questions:

Why not just combine @Async with a bulkhead semaphore based implementation?
Can both annotations be used together? If yes, why is the reason for a threadPool implementation?

@Async and @Bulkhead combined.

@Bulkhead(name = "Service3", fallbackMethod = "futureFallback")
    @Async
    public CompletableFuture<String> doSomeWork() {
        System.out.println("Excecuting service 3 - " + Thread.currentThread().getName());   
        Util.mockExternalServiceHttpCall(DELAY);
        return CompletableFuture.completedFuture("ok");
    }

Complete Code.

Yes, you can use both annotations together. They allow you to generate a limited number of async calls based on your bulkhead semaphore configuration.

However, there are some implications that are useful to know.

SimpleAsyncTaskExecutor.

When you annotate a method with the @Async annotation, Spring can use different implementations of the TaskExecutor interface.
By default, the framework uses the SimpleAsyncTaslExecutor.

This implementation will create a new thread each time the annotated method is invoked; this thread will not be reused.

The problem with this approach is that you will create a new thread even if the semaphore counter is 0 (bulkhead is full)

As you can see in the following stacktrace, the framework first creates a thread and then calls the bulkhead pattern which determines if there are available permissions to continue the execution. If the semaphore counter is cero, the bulkhead will reject the method execution.

Thread [_simpleAsyncTask1] (Suspended (breakpoint at line 18 in Service3))  
    Service3.doSomeWork() line: 18  
    Service3$$FastClassBySpringCGLIB$$6085f5a4.invoke(int, Object, Object[]) line: not available    
    MethodProxy.invoke(Object, Object[]) line: 218  
    CglibAopProxy$CglibMethodInvocation.invokeJoinpoint() line: 793 
    CglibAopProxy$CglibMethodInvocation(ReflectiveMethodInvocation).proceed() line: 163 
    CglibAopProxy$CglibMethodInvocation.proceed() line: 763 
    MethodInvocationProceedingJoinPoint.proceed() line: 89  
    BulkheadAspect.lambda$handleJoinPointCompletableFuture$0(ProceedingJoinPoint) line: 225 
    1457434357.get() line: not available    
    Bulkhead.lambda$decorateCompletionStage$1(Bulkhead, Supplier) line: 100 
    1251257755.get() line: not available    
    SemaphoreBulkhead(Bulkhead).executeCompletionStage(Supplier<CompletionStage<T>>) line: 557  
    BulkheadAspect.handleJoinPointCompletableFuture(ProceedingJoinPoint, Bulkhead) line: 223    
    BulkheadAspect.proceed(ProceedingJoinPoint, String, Bulkhead, Class<?>) line: 162   
    BulkheadAspect.lambda$bulkheadAroundAdvice$5eb13a26$1(ProceedingJoinPoint, String, Bulkhead, Class) line: 129   
    1746723773.apply() line: not available  
    1746723773(CheckedFunction0<R>).lambda$andThen$ca02ab3$1(CheckedFunction1) line: 265    
    1151489454.apply() line: not available  
    BulkheadAspect.executeFallBack(ProceedingJoinPoint, String, Method, CheckedFunction0<Object>) line: 139 

==> here
 BulkheadAspect.bulkheadAroundAdvice(ProceedingJoinPoint, Bulkhead) line: 128   
    NativeMethodAccessorImpl.invoke0(Method, Object, Object[]) line: not available [native method]  
    NativeMethodAccessorImpl.invoke(Object, Object[]) line: 62  
    DelegatingMethodAccessorImpl.invoke(Object, Object[]) line: 43  
    Method.invoke(Object, Object...) line: 566  
    AspectJAroundAdvice(AbstractAspectJAdvice).invokeAdviceMethodWithGivenArgs(Object[]) line: 634  
    AspectJAroundAdvice(AbstractAspectJAdvice).invokeAdviceMethod(JoinPoint, JoinPointMatch, Object, Throwable) line: 624   
    AspectJAroundAdvice.invoke(MethodInvocation) line: 72   
    CglibAopProxy$CglibMethodInvocation(ReflectiveMethodInvocation).proceed() line: 175 
    CglibAopProxy$CglibMethodInvocation.proceed() line: 763 
    ExposeInvocationInterceptor.invoke(MethodInvocation) line: 97   
    CglibAopProxy$CglibMethodInvocation(ReflectiveMethodInvocation).proceed() line: 186 
    CglibAopProxy$CglibMethodInvocation.proceed() line: 763 
    AnnotationAsyncExecutionInterceptor(AsyncExecutionInterceptor).lambda$invoke$0(MethodInvocation, Method) line: 115  
    1466446116.call() line: not available   
    AsyncExecutionAspectSupport.lambda$doSubmit$3(Callable) line: 278   
    409592088.get() line: not available 
    CompletableFuture$AsyncSupply<T>.run() line: 1700   

    ==> here
SimpleAsyncTaskExecutor$ConcurrencyThrottlingRunnable.run() line: 286   
    Thread.run() line: 829

After executing a load test that invokes for 60 seconds an @Aync and @Bulkhead annotated method, you can see from a profiling tool picture, that the application only used 34 out of 514 created threads. This obviously represents a waste of resources.

ThreadPoolTaskExecutor

Another option could be to use the TreadPoolTaskExecutor implementation.
After executing the same test using this implementation, the number of created threads decreased a lot (41).

However, the problem with this approach is that we are using unnecessary redundancy. Using a thread pool and a semaphore together, in my opinion, has no real advantage.

Conclusion.

For limiting asynchronous calls, use a Bulkhead threadPool implementation instead of a combination of @Async and Bulkhead semaphore.