【发布时间】:2021-05-04 03:59:24
【问题描述】:
很长一段时间以来,我一直在使用来自 kafka 主题的事件并在我的应用程序中处理相同的事件。 该主题有 20 个分区,我将 kafka 并发设置为 10 ,因为我使用我的应用程序的 2 个副本使用来自 kafka 主题的事件。我将提交模式设置为手动立即,因此一旦应用程序确保事件得到处理,我想提交分区的偏移量。一切都很好,直到有一天,一个或多个 kafka 服务器节点关闭并重新启动。我们使用 3 个 kafka 代理节点。发生这种情况,我碰巧看到 在消费者群体中发生了很多再平衡,它被踢出并在一段时间内不断加入。然后突然,我开始 注意到每个消费者(我的应用程序的每个副本中的 10 个线程组)按如下方式打印日志
"found no committed offset for partition"
"Resetting offset to {NUMBER} for the partition {TOPIC-NUMBER}"
在这个日志之后,每个消费者开始从每个分区读取最早可用的偏移量(它全部由应用程序在几天前提交)。这是正常行为吗?我试图调查这个问题。我发现的只是,以下
- KAFKA broker/server 存储每个 topic、partition 和 GroupNamee 组合的偏移量信息。
- 由于我使用手动提交调用确认(我使用的是 spring boot + kafka 技术堆栈),一旦我成功提交特定偏移量,偏移量就会增加。因此,该特定消费者的新偏移量将是committedOffset+1。
- 例如,如果 1 在特定主题分区从偏移量 0 开始消费,并且在运行应用程序 10 天后,它将达到 10K(用于计算目的),直到 kafka 代理重新启动/杀死,它正在读取10XXX 系列。有一次,它重新启动,我盯着 10 天前的 kafka 消费事件。这可能吗? .可以将所有这些已提交的偏移量视为未提交的偏移量吗?如果未提交,那么为什么我在 10 天后开始阅读事件?当我使用手动提交时偏移量是如何移动的?我一点头绪都没有
- 来自 Kafka 消费者 API (2.5.2),我的理解是当服务器响应特定分区的偏移量为 -1 时,消费者使用配置的重置策略重置偏移量(在我的情况下最早)。很明显,我在 kafka 服务器重新启动期间收到了 -1。但我不知道这可能是什么。
我的kafka客户端属性如下
ofset.reset= earliest ( but it should not provide me committed offsets again, I guess)
heat beat interval = 2000 ms
session time out= 50000 ms
auto commit = false
我注意到之前在这里问过类似的问题。但没有什么能解决这个问题。我在这里提到了这些链接
Kafka-node suddenly consumes from offset 0
Kafka consumer: starts reading partition from the beginning even thought there's a committed offset
After kafka crashed, the offsets are lost
Kafka partitions and offsets disappeared
【问题讨论】:
-
重启集群不应修改
__consumer_offsets主题。但是,如果您根本没有提交偏移量(您应该向您的代码展示您是如何执行此操作的),那么消费者组从一开始就不存在,它将执行偏移量重置。根据您的分段文件的大小,完全有可能使用 10 天前的数据 -
您的代理是什么版本的?在 2.0 之前,即使消费者仍处于连接状态,代理也会在 24 小时内没有消费任何内容时使偏移量过期。在 2.0 中,改为 7 天。在 2.1(我认为)中,它被更改为仅在消费者 7 天未连接时才过期。请参阅kafka.apache.org/documentation/#upgrade_200_notable 和
标签: java apache-kafka kafka-consumer-api