如果 Kafka 的消费者处理消息的时间过长怎么办？ Kafka 是否会将此分区重新指定给另一个消费者，并且消息会被双重处理？

Question

假设Kafka、1 partition、2 consumers。（第二个消费者空闲）

假设第一个消费了一条消息，然后与其他 3 个服务一起处理它，然后突然卡在其中一个上并错过了 Kafka 的超时。

Kafka 是否会将分区重新分配给第二个消费者，并且消息将被双重处理（假设第一个最终成功）？

Answer 1

如果 Kafka 的消费者处理消息的时间过长怎么办？ Kafka 是否会将此分区重新指定给另一个消费者，并且消息会被双重处理？

是的，没错。如果 Kafka 消费者处理消息的时间过长并且后续 poll() 被延迟，Kafka 将重新分配这个分区给另一个消费者，并且消息将被再次处理（一次又一次）。

为了更清楚，首先我们需要决定并定义“多长是太长？”。

这是由属性max.poll.interval.ms 定义的。来自docs，

使用消费者组管理时调用 poll() 之间的最大延迟。这为消费者在获取更多记录之前可以空闲的时间量设置了上限。如果在此超时到期之前未调用 poll()，则认为消费者失败，组将重新平衡，以便将分区重新分配给另一个成员。

如果在这段时间内没有调用 poll()，则重新平衡消费者组。

还有一个属性auto.commit.interval.ms。只有在轮询期间才会调用自动提交偏移量检查 - 它检查经过的时间是否大于配置的自动提交间隔时间，如果结果为是，则提交偏移量。

如果 Kafka 消费者处理记录的时间过长，那么随后的 poll() 调用也会延迟，并且不会提交最后一次 poll() 返回的偏移量。如果此时发生再平衡，分配给该分区的新消费者客户端将重新开始处理消息。

可以通过增加此值来避免消费者组重新平衡和由此产生的分区重新分配。这将增加轮询之间的允许间隔，并为消费者提供更多时间来处理从 poll() 返回的记录。消费者只会在轮询调用中加入重新平衡，因此增加最大轮询间隔也会延迟组重新平衡。

将最大轮询间隔增加到一个大值还有一个问题。如果消费者因其他原因死亡，检测到故障的时间将超过配置的max.poll.interval.ms 间隔。

session.timeout.ms 和heartbeat.interval.ms 在这种情况下可用，以便尽早检测到完全故障。

有关这些参数的更多详细信息：

• 请参考this
• KIP-62

请注意，为session.timeout.ms 配置的值必须在代理配置中通过属性配置的允许范围内

• group.min.session.timeout.ms
• group.max.session.timeout.ms

否则，启动消费者客户端时会抛出以下异常。

Exception in thread "main" org.apache.kafka.common.errors.InvalidSessionTimeoutException:
The session timeout is not within the range allowed by the broker
(as configured by group.min.session.timeout.ms and group.max.session.timeout.ms)

更新：避免再次处理消息

KafkaConsumer 类commitAsync() 中还有另一个方法可以触发提交偏移操作。

ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(500));
kafkaConsumer.commitAsync();

关于 commitSync() 和 commitAsync() 的更多详细信息，请查看this thread

手动提交偏移量是表示该偏移量已被处理，这样 Kafka 就不会再次发送同一分区的已提交记录。当手动提交偏移量时，请务必注意，如果消费者在处理记录之前因任何原因死亡，则这些记录可能不会再次被处理。