Circuit breaker implementation for Kafka Consumer

This solution is implemented using Resilience4j for circuit breaker.

I have Kafka consumer implemented using spring-cloud-stream. Consumer code connects to external HTTP service( this can be any external/internal dependent service). When such service is down, you want to stop consuming messages from Queue/Topic. The Circuit Breaker is configured on this layer. As multiple messages start failing when external HTTP service is down then circuit breaker gets OPEN(circuit breaker between Consumer Service & External Service) and on state transition(CLOSE to OPEN/ HALF_OPEN ) of this circuit breaker, stop consumer. so no more messages will be processed until circuit is closed again.

Using this approach your messages will stay on queue/topic and will be processed again whenever circuit gets CLOSE. No need of processing messages from error queue/dead letter later.

Consumer – spring-cloud-stream consumer configured to connect to Kafka Topic.
Service – consumer calls service for processing message read from Topic. This service calls HTTPClient service for connecting to external service
HTTP Client – can be service having REST client which calls external service

Steps for circuit breaker implementation for

1. Step 1: Configure resilience4J for circuit breaker on HttpClientService.
   1. Circuit breaker on HTTP client opens when external service throws for example HttpServerErrorException (for any 5XX errors)
   2. Resilience4J will first retry the message as per configured retry frequency
   3. When said number of calls to external service are failed then Resilence4J opens circuit between Service & HTTP Client.
2. Step 2: Implement Circuit Breaker state transition event.
   1. When circuit breaker state is OPEN then stop consumer. Use org.springframework.cloud.stream.endpoint.BindingsEndpoint to STOP or START consumer
   2. When circuit breaker is HALF OPEN after configured wait duration then start consumer again

Following code snippet uses resilience4J api for circuit breaker implementation & spring-cloud-stream for consumer implementation –
implementation ‘io.github.resilience4j:resilience4j-spring-boot2:1.2.0’
implementation ‘org.springframework.cloud:spring-cloud-stream’
implementation ‘org.springframework.cloud:spring-cloud-stream-binder-kafka’
implementation ‘org.springframework.kafka:spring-kafka’

Circuit Breaker Properties –
Each call to external service will be retried 3 times and if percent(%) failure in five consecutive calls is more than 80% then circuit breaker will OPEN and on next call to external service will be blocked (circuit breaker will throw a CallNotPermittedException)

Circuit Breaker Implementation for HttpClient

Consumer with circuit breaker state transition Event

Configure method on state transition of circuit breaker
this.circuitBreaker.getEventPublisher().onStateTransition(this::onStateChange);
This “onStateChange” method will stop/start consumer. See code snippet below –

When circuit breaker is OPEN call bindingsEnpoint.changeState() to stop consumer. But consumer doesn’t stop immediately until all messages which are already polled(if poll size is greater than 1) are processed. So to discard these rest of polled messages call acknowledgment.nack(). nack() method is implemented for Kafka so this may not work for other types of Queues/Topics. So for others you can set poll size to 1 (which might not be feasible solution as you don’t want to connect to queue frequently).

Be yourself; Everyone else is already taken.

— Oscar Wilde.

Stay tuned for more. Subscribe below to get notified when I post new updates. You can post suggestions or questions for any other content you are looking about Kafka.