Spring Cloud Stream Kafka retries 10 times the maxAttempts

user_9840916 0
Viewed 1932

I've been trying to implement retry logic for Spring cloud stream kafka such that if an exception is throw when producing an event to the topic sample-topic, It retries two more time.
I added in the following configuration to the application.properties file

spring.cloud.stream.bindings.processSampleEvent.destination=sample-topic
spring.cloud.stream.bindings.processSampleEvent.content-type=application/json
spring.cloud.stream.bindings.processSampleEvent.consumer.maxAttempts=2

I've written the lister code in way that it simply logs the received message and throws a NullPointerException so that I can test out the retry.

@StreamListener(ListenerBind.SAMPLE_CHANNEL)
  public void processSampleEvent(String productEventDto) {
    System.out.println("Entering listener: " + productEventDto);
    throw new NullPointerException();
}

But when testing out by producing an event to the sample-topic, I see that in the logs the event has been retries 20 times but I've specified in the properties to try only two time and also a weird thing happens when I change to it 3 times, It retries 30 times.
I'm pretty new to Spring cloud streams and any help on this would be really helpful. Thanks in Advance

spring-kafka spring-cloud-stream-binder-kafka spring-cloud-stream spring-boot
2 Answers

The default error handler in the listener container is now a SeekToCurrentErrorHandler with 10 delivery attempts.

You can either disable the retries in the binder, and configure a STCEH with the retry semantics you want, or use retries in the binder and replace the default error handler with a simple LoggingErrorHandler.

To configure the container's error handler, add a ListenerContainerCustomizer<AbstractKafkaListenerContainerFactory> @Bean.

user_12407630

I faced the same problem. My working solution was to create a ListenerContainerCustomizer Bean, give it desired number of max attempts, and set consumer binding maxAttempts: 1

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?,?>> listenerContainerCustomizer(){
    return (container, dest, group) ->
          container.setErrorHandler(containerAwareErrorHandler());
}
public SeekToCurrentErrorHandler containerAwareErrorHandler(){
    return new SeekToCurrentErrorHandler(new FixedBackOff(0, maxAttempts-1);
}
user_92214310
Related