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

Question

我想通过属性为Spring Cloud Stream ；在Kafka consumer上设置一个自动偏移提交间隔。

正如我从度量中看到的，默认情况下，Spring Cloud Stream ；Kafka对每个消耗的消息提交偏移量。对于高负载的主题（例如，如果流量为每秒10K消息），它会变得非常明显，并增加Kafka broker的负载。

我们以以下方式声明消费者：

@Bean
public Consumer<TestEvent> testEvents() {
    …
}

我尝试了几个选择，但都没有帮助我。

spring:
  cloud:
    function:
      definition: testEvents
    stream:
      kafka:
        binder:
          brokers: xxx
          consumer-properties:
            auto.offset.reset: latest
            auto.commit.interval.ms: 2000
            enable.auto.commit: true
      bindings:
        testEvents-in-0:
          destination: test_topic
          group: ${spring.application.name}_test_topic
          consumer:
            concurrency: 2

{
  "status": "DOWN",
  "components": {
    "binders": {
      "status": "DOWN",
      "components": {
        "kafka": {
          "status": "DOWN",
          "details": {
            "error": "org.apache.kafka.common.KafkaException: Failed to construct kafka consumer"
          }
        }
      }
    }, …

据我所知，enable.auto.commit:true不建议用于Spring Cloud Stream。

备选方案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
            ack-mode: TIME
            ack-time: 2000

但是，这样的配置属性ack-mode:time并没有帮助，因为我们看到每个消耗的消息都有偏移量提交。

我使用maven dependencyspring-cloud-starter-stream-kafkaversion3.0.12.release

Answer 1

首先，ack-mode是一个特定于Kafka的使用者绑定属性，您在公共属性中有它。

spring:
  cloud:
    function:
      definition: testEvents
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          configuration:
            auto.offset.reset: latest
        bindings:
          testEvents-in-0:
            consumer:
              ack-mode: TIME
      bindings:
        testEvents-in-0:
          destination: test_topic
          group: ${spring.application.name}_test_topic
          consumer:
            concurrency: 2

其次，没有ACK-TIME属性，您必须通过容器定制器在容器上设置它：

@Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> cust() {
    return (container, dest, group) -> {
        container.getContainerProperties().setAckTime(2000L);
        container.getContainerProperties().setLogContainerConfig(true);
    };
}