如何从应用程序配置Kafka consumer retry属性。spring boot中的属性？

Question

在Spring靴中，

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

通过使用上述配置，我们可以避免在spring boot中基于java web（bean）的配置，这是一个很有价值的优势。

Q： 我们可以从应用程序中添加Kafka错误处理程序和Kafka消费者重试次数属性吗。属性/应用。yml？

我找不到任何关于它的参考或留档，因此希望有一些结论，只是因为这个问题，现在我必须在Spring Boot中转到java基于Web的配置并删除属性配置，这再次回到Spring中的旧方式。我相信应该有一些解决方法，我们可以通过属性文件配置来实现这一点。

Answer 1

消费者没有重试属性。如果未提交偏移量，下一次轮询将从相同的偏移量重试。

也没有任何可配置的错误处理类像Kafka Streams中那样脱离反序列化带外。

如果你想处理反序列化错误，而不是处理错误，你可以这样设置

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

除此之外，您还可以利用代码中的死信发布恢复器（DeadLetterPublishingRecoverer）和SeekToCurrentErrorHandler，将错误事件生成到新主题，以供检查和进一步处理来源