apache-kafka - Kafka增加处理超时消息

Question

在kafka消费者中，如果消息处理时间超过5分钟，则消息是新处理的，我已经配置消费者增加“max.poll.interval.ms”和session.timeout.ms

max.poll.interval.ms= 7200000 (2Hrs) session.timeout.ms= 7200000 (2Hrs) request.timeout.ms=7206000 (~2Hrs)

这起初有效，之后，它正在重新平衡
我在消费者身上遇到了这个错误

2019-10-18 15:29:51.739 INFO  o.a.k.c.c.i.AbstractCoordinator - [Consumer clientId=consumer-5, groupId=processgroup] (Re-)joining

group

Answer 1

你是在正确的方法。您还需要设置 session.timeout.ms 在消费者和
group.max.session.timeout.ms 在 Kafka 经纪人中以避免重新平衡。

After subscribing to a set of topics, the consumer will automaticallyjoin the group when poll(long) is invoked. The poll API is designed toensure consumer liveness. As long as you continue to call a poll, theconsumer will stay in the group and continue to receive messages fromthe partitions it was assigned. Underneath the covers, the consumersends periodic heartbeats to the server. If the consumer crashes or isunable to send heartbeats for a duration of session.timeout.ms, thenthe consumer will be considered dead and its partitions will bereassigned.

max.poll.interval.ms: 使用消费者组管理时调用 poll() 之间的最大延迟。如果在此超时到期之前没有调用 poll()，则认为消费者失败，组将重新平衡以将分区重新分配给另一个成员。

Note: Please note if the max poll increased a lot it will delay a grouprebalance because consumer rebalance to join only when poll getcalled.

request.timeout.ms: 配置控制客户端等待请求响应的最长时间。如果在超时之前没有收到响应，客户端将在必要时重新发送请求，或者如果重试用尽，则请求失败。

Note: This indirectly not impact on rebalancing however it need to set becauseit must always be larger than max.poll.interval.ms else it throwsconfiguration error.

session.timeout.ms: 使用 Kafka 的组管理工具时用于检测消费者故障的超时时间。消费者定期发送心跳以向代理表明其活跃度。如果在此 session 超时到期之前代理没有收到心跳，则代理将从组中删除此消费者并启动重新平衡。

Note: But setting session.timeout.ms little tricky it alone would notwork you need to check Kafka broker settinggroup.min.session.timeout.ms that also need to increase.

group.max.session.timeout.ms(在 Kafka 代理中) :已注册消费者允许的最大 session 超时时间。其默认值为 30000 毫秒。