如何通过配置属性设置Spring Cloud Stream Kafka的偏移量提交间隔

失败原因说明 #

• 方案1失败原因：开启enable.auto.commit=true后，Kafka原生自动提交逻辑会和Spring Cloud Stream内置的偏移量管理、重试、死信队列等特性冲突，导致健康检查时无法正常构造消费者实例，最终返回健康状态DOWN。框架官方也不推荐使用Kafka原生自动提交模式。
• 方案2失败原因：ack-mode、ack-time属于Kafka binder特有的配置项，不能配置在通用绑定的spring.cloud.stream.bindings.<绑定名称>.consumer节点下，必须放置到Kafka专属的绑定配置节点才会被加载生效。

正确配置示例 #

针对你使用的3.0.12.RELEASE版本，使用如下配置即可实现每2秒批量提交一次偏移量：

spring:
  cloud:
    function:
      definition: testEvents
    stream:
      kafka:
        binder:
          brokers: xxx
          configuration:
            auto.offset.reset: latest
      bindings:
        testEvents-in-0:
          destination: test_topic
          group: ${spring.application.name}_test_topic
          consumer:
            concurrency: 2
      # Kafka专属绑定配置节点
      kafka:
        bindings:
          testEvents-in-0:
            consumer:
              # 按时间间隔提交偏移量
              ack-mode: TIME
              # 提交间隔，单位为毫秒
              ack-time: 2000

如果需要对所有消费者全局生效间隔提交规则，可以添加全局默认配置：

spring:
  cloud:
    stream:
      kafka:
        default:
          consumer:
            ack-mode: TIME
            ack-time: 2000

注意事项 #

按时间间隔提交偏移量的模式下，若服务发生异常宕机，最后一次提交后到宕机前消费的消息偏移量未及时持久化，重启后会出现少量重复消费的情况，需要确保业务消费逻辑具备幂等性。

内容的提问来源于stack exchange，提问作者Vasyl Sarzhynskyi