best way to use circuitbreaker in kafka listener

[fuesec]

I have a spring boot application written in java which has a kafka listener:

    @KafkaListener(
            autoStartup = "false",
            topics = "topic name",
            containerFactory = "byteArrayKafkaListenerContainerFactory",
            concurrency = "8",
            batch = "true",
            clientIdPrefix = "clientIdPrefix",
            id = "#{__listener.id}")
    @Override
    public void accept(List<Message<byte[]>> messages) {
        circuitBreaker
                .decorateRunnable(() -> retryTemplate.execute(context -> processMessages(messages)))
                .run();
    }

The circuit breaker (resilience4j) configuration is highlighted below. The circuitBreaker which is being used above is the last thing being configured in the code-block below.

    @Bean
    CircuitBreakerConfig circuitBreakerConfig() {
        return CircuitBreakerConfig.custom()
                .failureRateThreshold(50)
                .minimumNumberOfCalls(1)
                .waitDurationInOpenState(Duration.ofMinutes(1))
                .slidingWindowSize(1)
                .build();
    }

    @Bean
    CircuitBreakerRegistry circuitBreakerRegistry(CircuitBreakerConfig circuitBreakerConfig) {
        return CircuitBreakerRegistry.of(circuitBreakerConfig);
    }

    @Bean
    CircuitBreaker circuitBreaker(
            CircuitBreakerRegistry circuitBreakerRegistry,
            KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry) {
        var circuitBreaker = circuitBreakerRegistry.circuitBreaker("circuitBreakerId");
        circuitBreaker.getEventPublisher().onStateTransition(event -> {
            var stateTransition = event.getStateTransition();
            var listenerContainerId = "listenerId";
            switch (stateTransition) {
                case CLOSED_TO_OPEN, CLOSED_TO_FORCED_OPEN, HALF_OPEN_TO_OPEN: {
                    var container = kafkaListenerEndpointRegistry.getListenerContainer(listenerContainerId);
                    if (container != null) {
                        container.pause();
                    }
                    break;
                }
                case OPEN_TO_HALF_OPEN, HALF_OPEN_TO_CLOSED, FORCED_OPEN_TO_CLOSED, FORCED_OPEN_TO_HALF_OPEN: {
                    var container = kafkaListenerEndpointRegistry.getListenerContainer(listenerContainerId);
                    if (container != null) {
                        container.resume();
                    }
                    break;
                }
                default:
                    throw new IllegalStateException("Unknown state transition: " + stateTransition);
            }
        });
        return circuitBreaker;
    }

If the circuit breaker encounters the first exception the event publisher onStateTransition code is triggered and therefore pausing is requested for the kafka listener container. The next thing I see in the logs is that all kafka partitions are revoked and partitions are assigned again. In between the partition assignment messages I see the following:

....KafkaMessageListenerContainer    : Paused consumer resumed by Kafka due to rebalance; consumer paused again, so the initial poll() will never return any records

The problem is the kafka listener container is no longer paused but also does not resume listening to the topic. I would like to properly pause the kafka listener container and resume it once the circuit breaker transitions to a corresponding state. What is the best way to accomplish this? I want to accomplish a pausing of the kafka listener without causing a rebalance. Scenario: two pods of this spring boot java application running in a kubernetes cluster and the kafka listener of one of those pods runs into temporary problem processing messages (e.g. persisting to database due to connection issue) and therefore should stop processing messages until the circuit breaker deems it ok to resume.

stackoverflow question: https://stackoverflow.com/questions/79185634/best-way-to-use-circuitbreaker-in-kafka-listener

[artembilan]

Please, don't repeat questions in different forums which are indeed monitored by the team.