京公网安备11010502063487号CommitFailedException 说“消费者被踢出了组”,但原因几乎总是“处理一批消息花的时间超过了 max.poll.interval.ms”。这篇用 Kafka 4.3.1 在本机复现它:看日志和异常的原文、重复消费是怎么发生的、三种修法的区别,并发现新消费组协议下同样的超时不抛异常。
classic_fail(max.poll.records=5,max.poll.interval.ms=3000,每条处理 1000 ms,group.protocol=classic):
[ 5973 ms] 第 1 次 poll 拿到 5 条,第一条 offset = 0
LOG WARN ConsumerCoordinator: [Consumer clientId=consumer-g_classic_fail-1, groupId=g_classic_fail] 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.
[11001 ms] 处理完 5 条(每条 1000 ms),准备 commitSync
[11002 ms] commitSync 失败:org.apache.kafka.clients.consumer.CommitFailedException
message: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.
[11013 ms] 第 2 次 poll 拿到 5 条,第一条 offset = 0
LOG WARN ConsumerCoordinator: [Consumer clientId=consumer-g_classic_fail-1, groupId=g_classic_fail] 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.
[16048 ms] 处理完 5 条(每条 1000 ms),准备 commitSync
[16048 ms] commitSync 失败:org.apache.kafka.clients.consumer.CommitFailedException
message: Offset commit cannot be completed since the consumer is not part of an active group for auto partition assignment; it is likely that the consumer was kicked out of the group.
最后查询已提交 offset:[null]两次 poll 的间隔超过 max.poll.interval.ms,commitSync 抛 CommitFailedException,重新 poll 重复消费。
fix_records(max.poll.records=2,max.poll.interval.ms=3000,每条处理 1000 ms,group.protocol=classic):
[ 5462 ms] 第 1 次 poll 拿到 2 条,第一条 offset = 0
[ 7477 ms] 处理完 2 条(每条 1000 ms),准备 commitSync
[ 7487 ms] commitSync 成功
[ 7488 ms] 第 2 次 poll 拿到 2 条,第一条 offset = 2
[ 9506 ms] 处理完 2 条(每条 1000 ms),准备 commitSync
[ 9507 ms] commitSync 成功
最后查询已提交 offset:[OffsetAndMetadata{offset=4, leaderEpoch=0, metadata=''}]
fix_interval(max.poll.records=5,max.poll.interval.ms=10000,每条处理 1000 ms,group.protocol=classic):
[ 5484 ms] 第 1 次 poll 拿到 5 条,第一条 offset = 0
[10517 ms] 处理完 5 条(每条 1000 ms),准备 commitSync
[10520 ms] commitSync 成功
[10520 ms] 第 2 次 poll 拿到 5 条,第一条 offset = 5
[15546 ms] 处理完 5 条(每条 1000 ms),准备 commitSync
[15547 ms] commitSync 成功
最后查询已提交 offset:[OffsetAndMetadata{offset=10, leaderEpoch=0, metadata=''}]
fix_speed(max.poll.records=5,max.poll.interval.ms=3000,每条处理 300 ms,group.protocol=classic):
[ 5458 ms] 第 1 次 poll 拿到 5 条,第一条 offset = 0
[ 6973 ms] 处理完 5 条(每条 300 ms),准备 commitSync
[ 6984 ms] commitSync 成功
[ 6985 ms] 第 2 次 poll 拿到 5 条,第一条 offset = 5
[ 8519 ms] 处理完 5 条(每条 300 ms),准备 commitSync
[ 8521 ms] commitSync 成功
最后查询已提交 offset:[OffsetAndMetadata{offset=10, leaderEpoch=0, metadata=''}]减小 max.poll.records、调大 max.poll.interval.ms、加快处理,本质是让一批的处理时间小于间隔。
new_fail(max.poll.records=5,max.poll.interval.ms=3000,每条处理 1000 ms,group.protocol=consumer):
[ 5553 ms] 第 1 次 poll 拿到 5 条,第一条 offset = 0
LOG WARN ConsumerHeartbeatRequestManager: [Consumer clientId=consumer-g_new_fail-1, groupId=g_new_fail] 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.
[10593 ms] 处理完 5 条(每条 1000 ms),准备 commitSync
[10594 ms] commitSync 成功
LOG WARN ConsumerHeartbeatRequestManager: [Consumer clientId=consumer-g_new_fail-1, groupId=g_new_fail] Time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, exceeded approximately by 2067 ms. Member vNfjoi4BSZGdzBgnrK5TKg will rejoin the group now.
[10820 ms] 第 2 次 poll 拿到 5 条,第一条 offset = 0
LOG WARN ConsumerHeartbeatRequestManager: [Consumer clientId=consumer-g_new_fail-1, groupId=g_new_fail] 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.
[15855 ms] 处理完 5 条(每条 1000 ms),准备 commitSync
[15856 ms] commitSync 成功
最后查询已提交 offset:[]group.protocol=consumer 时 commitSync 返回成功但没有提交,不抛异常。
消费组 g_classic_fail:
(没有已提交的 offset)
消费组 g_new_fail:
(没有已提交的 offset)
消费组 g_fix_records:
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
g_fix_records orders 0 4 30 26 - - -用 kafka-consumer-groups --describe 看提交点,不要只信客户端。