Spring Cloud Stream Kafka3.0.4报CommitFailedException异常咨询

错误日志 #

org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:900)
        at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator$OffsetCommitResponseHandler.handle(ConsumerCoordinator.java:840)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:978)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:958)
        at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:204)
        at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:167)
        at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:127)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.fireCompletion(ConsumerNetworkClient.java:578)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.firePendingCompletedRequests(ConsumerNetworkClient.java:388)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:294)
        at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:303)
        at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$HeartbeatThread.run(AbstractCoordinator.java:1104)

根因分析 #

该异常是Kafka消费组重平衡导致的偏移量提交失败，核心触发逻辑如下：

1. Kafka消费者调用poll()拉取到一批消息后，必须在max.poll.interval.ms指定的时间窗口内完成整批消息处理，再次发起下一次poll()调用。
2. 如果两次poll的间隔超过配置阈值，Kafka服务端会判定当前消费者处理能力不足/已失活，立即触发消费组重平衡，将该消费者持有的分区所有权转移给组内其他存活消费者。
3. 重平衡完成后，原消费者再尝试提交之前拉取批次的消息偏移量时，因为已经丧失对应分区的消费权限，就会抛出CommitFailedException。

从当前配置看，未显式配置max.poll.interval.ms和max.poll.records，会直接使用Kafka客户端默认值：max.poll.interval.ms=300000（5分钟）、max.poll.records=500（单次拉取500条消息）。如果单条消息处理耗时较长，或者消费逻辑存在阻塞，500条消息的总处理时长很容易突破5分钟阈值，触发重平衡。
注意当前配置的auto.commit.interval.ms=1000无法规避该问题：偏移量自动提交动作嵌入在poll()调用流程中执行，一旦消费线程卡在消息处理阶段无法发起poll，自动提交根本不会触发；就算后台心跳线程正常运行，只要超过max.poll.interval.ms未发起poll，消费者一样会被踢出消费组。

修复方案 #

优先排查代码逻辑阻塞点，再配合参数调整适配业务处理耗时：

配置调整 #

在spring.cloud.stream.kafka.binder.consumer-properties配置段下新增以下参数，具体数值结合实际业务压测结果调整：

spring:
   cloud: 
     stream:  
       bindings:
         default:
           content-type: application/*+avro
         inputAddAccount:
           destination: dev_account
           group: dev_group
         # 修正原配置的拼写错误：ouput -> output，避免后续通道绑定失效
         outputUpdateFile:
           destination: dev_update_file
           group: dev_group
           producer:
            useNativeEncoding: true
       kafka:
         binder:
           brokers: kafka1-dev:6667
           auto-create-topics: false
           consumer-properties:
             auto.offset.reset: latest
             auto.commit.interval.ms: 1000
             specific.avro.reader: true
             key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
             value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
             schema.registry.url: https://schema-registry1.dev:8088
             basic.auth.credentials.source: USER_INFO
             basic.auth.user.info: reader:reader
             # 新增消费者核心参数
             max.poll.interval.ms: 600000 # 单批消息最大处理时长，设为10分钟，根据实际业务耗时留20%-30%冗余
             max.poll.records: 100 # 单次poll拉取的消息条数，从默认500降到100，缩短单批总处理时长
           producer-properties:
             acks: -1
             retries: 2147483647
             max.in.flight.requests.per.connection : 1
             request.timeout.ms: 10000
             key.serializer: org.apache.kafka.common.serialization.StringSerializer
             value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
             schema.registry.url: https://schema-registry1.dev:8088
             basic.auth.credentials.source: USER_INFO
             basic.auth.user.info: reader:reader

代码优化 #

• 排查消费逻辑阻塞点：检查消费方法里是否存在无超时的外部接口调用、慢SQL、大文件同步IO、不必要的线程等待等长耗时操作，给所有远程调用、IO操作设置合理超时，尽量压缩单条消息的处理耗时。
• 如果业务确实存在无法缩短的长耗时处理逻辑，不要在Kafka消费线程里同步执行处理，可将消息转发到自定义业务线程池异步处理，此时需要关闭自动提交，等业务逻辑实际处理完成后再手动提交偏移量，避免消息丢失。
• 消费逻辑要做好异常捕获，不要让未捕获的RuntimeException卡住消费线程，导致无法发起下一次poll调用。

内容的提问来源于stack exchange，提问作者Aymen Kanzari