关于Kafka监听器容器超max.poll.interval.ms后仍处理记录的行为问询

我们检测到疑似异常行为：消费者从单分区主题拉取10条记录，逐条处理时在第3条超出max.poll.interval.ms，Kafka Broker断开该消费者及所属消费组连接，但监听器容器仍继续处理完所有10条记录。若存在同组多应用实例，另一实例会获取该分区并重复处理相同记录，引发重复消费。处理完成后消费者尝试提交偏移量，却因已退出消费组而失败。

使用版本：Spring Boot 3.0.4 + Spring Cloud 2022.0.2

超出max.poll.interval.ms前的日志 #

2023-03-13T09:23:11.413+01:00  INFO 10760 --- [container-0-C-1] o.a.k.c.c.internals.SubscriptionState    : [Consumer clientId=consumer-my-group-2, groupId=my-group] Resetting offset for partition topic1-0 to position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[127.0.0.1:9092 (id: 1001 rack: null)], epoch=0}}.
2023-03-13T09:23:11.414+01:00  INFO 10760 --- [container-0-C-1] o.s.c.s.b.k.KafkaMessageChannelBinder$2  : my-group: partitions assigned: [topic1-0]
2023-03-13T09:23:11.424+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : [topic1-0@0, topic1-0@1, topic1-0@2, topic1-0@3, topic1-0@4, topic1-0@5, topic1-0@6, topic1-0@7, topic1-0@8, topic1-0@9]
2023-03-13T09:23:11.425+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@0
2023-03-13T09:23:21.441+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@1
2023-03-13T09:23:31.443+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@2
2023-03-13T09:23:31.480+01:00  WARN 10760 --- [read | my-group] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-2, groupId=my-group] consumer poll timeout has expired. This means 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 processing messages. 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.

断开连接后继续处理的日志 #

此时处理第3条记录已超出max.poll.interval.ms，Kafka Broker断开消费者及组连接，但消费者仍继续处理所有记录：

2023-03-13T09:23:41.444+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@3
2023-03-13T09:23:51.445+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@4
2023-03-13T09:24:01.447+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@5
2023-03-13T09:24:11.449+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@6
2023-03-13T09:24:21.450+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@7
2023-03-13T09:24:31.451+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@8
2023-03-13T09:24:41.453+01:00 TRACE 10760 --- [container-0-C-1] o.s.k.l.KafkaMessageListenerContainer    : Processing topic1-0@9

偏移量提交失败的日志 #

处理完成后尝试提交偏移量失败：

2023-03-13T09:24:51.460+01:00  INFO 10760 --- [container-0-C-1] o.a.k.c.c.internals.ConsumerCoordinator  : [Consumer clientId=consumer-my-group-2, groupId=my-group] Failing OffsetCommit request since the consumer is not part of an active group

该行为属于预期行为，而非异常，核心逻辑如下：

• Kafka Broker检测到消费者超出max.poll.interval.ms未发起poll请求时，会将其移出消费组并触发分区重平衡，但Broker不会主动通知消费者停止处理已拉取的消息。
• Spring Kafka监听器容器在拉取到一批消息后，会独立完成这批消息的处理流程，不会因为消费者被移出消费组而中断正在进行的处理。
• 偏移量提交失败是因为消费者已不在活跃消费组中，完全符合Kafka协议规则。

但该行为会引发重复消费风险：当同组其他实例接管分区后，会从上次提交的偏移量开始重新消费，导致已处理但未提交的消息被重复处理。

• 调小max.poll.records：减少单次拉取的消息数量，避免因处理时间过长超出阈值。
• 增大max.poll.interval.ms：如果单条消息处理时间确实较长，适当延长该阈值，给足处理时间。
• 异步化处理逻辑：将消息处理逻辑异步执行，让poll请求能按时发起，维持消费者在消费组中的活跃状态。

内容的提问来源于Stack Exchange，提问作者Fernando Blanch