How to configure Kafka consumer retry property from application.properties in spring boot?

user_11243901 0
Viewed 708

In spring-boot,

application.yml:

kafka:
   bootstrap-servers: localhost:9092
   listener:
    concurrency: 10
    ack-mode: MANUAL
   producer:
    topic: test-record
    key-serializer: org.apache.kafka.common.serialization.StringSerializer
    value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    retries: 3
    orn-record:
      timeout: 3
    #acks: 1
   consumer:
    groupId: test-record
    topic: test
    enable-auto-commit: false
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

By using above configuration, We can avoid java web(bean) based configuration in spring boot and it's a high worthy advantage.

Q: Can we add kafka error handler and kafka consumer number of retry properties from application.properties / application.yml ?

I could not find any reference or documentation about it hence hoping some conclusion, just because of this issue now I have to go to java web based configuration in spring boot and remove the properties configuration which is again going back to old way in spring. I believe there should be some workaround and we could achieve this through property file configuration.

spring-kafka kafka-consumer-api apache-kafka spring-boot spring
1 Answers

Consumers don't have a retry property. If the offsets were not committed, the next poll will try again from the same offsets.

There is also not any configurable error handling class that is out-of-band from deserialization like there is in Kafka Streams.

If you want to handle deserialization errors, and not processing errors, you can set that up like so

spring:
  kafka:
    bootstrap-servers: ...
    consumer:
      # Configures the Spring Kafka ErrorHandlingDeserializer that delegates to the 'real' deserializers
      key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
    properties:
      # Delegate deserializers
      spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
      spring.deserializer.value.delegate.class: org.apache.kafka.common.serialization.StringDeserializer

Beyond that, you can leverage DeadLetterPublishingRecoverer and SeekToCurrentErrorHandler in the code to produce the error events to a new topic for inspection and further processing. - Source

user_23086830
Related