Apache Kafka 消费者组的偏移量如何到期? [英] How does an offset expire for an Apache Kafka consumer group?
问题描述
当我注意到一些奇怪的行为时，我正在对一个旧主题进行一些测试.阅读 Kafka 的日志，我注意到这个removed 8 expired offsets"消息:
I was making some tests on an old topic when I noticed some strange behaviours. Reading Kafka's log I noticed this "removed 8 expired offsets" message:
[GroupCoordinator 1001]: Stabilized group GROUP_NAME generation 37 (kafka.coordinator.GroupCoordinator)
[GroupCoordinator 1001]: Assignment received from leader for group GROUP_NAME for generation 37 (kafka.coordinator.GroupCoordinator)
Deleting segment 0 from log __consumer_offsets-31. (kafka.log.Log)
Deleting segment 0 from log __consumer_offsets-45. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-45/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting index /data/kafka-logs/__consumer_offsets-31/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting segment 0 from log __consumer_offsets-13. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-13/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting segment 0 from log __consumer_offsets-11. (kafka.log.Log)
Deleting segment 4885 from log __consumer_offsets-11. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-11/00000000000000004885.index.deleted (kafka.log.OffsetIndex)
Deleting index /data/kafka-logs/__consumer_offsets-11/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting segment 0 from log __consumer_offsets-26. (kafka.log.Log)
Deleting segment 12406 from log __consumer_offsets-26. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-26/00000000000000012406.index.deleted (kafka.log.OffsetIndex)
Deleting index /data/kafka-logs/__consumer_offsets-26/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting segment 0 from log __consumer_offsets-22. (kafka.log.Log)
Deleting segment 8643 from log __consumer_offsets-22. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-22/00000000000000008643.index.deleted (kafka.log.OffsetIndex)
Deleting index /data/kafka-logs/__consumer_offsets-22/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting segment 0 from log __consumer_offsets-6. (kafka.log.Log)
Deleting segment 9757 from log __consumer_offsets-6. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-6/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
Deleting index /data/kafka-logs/__consumer_offsets-6/00000000000000009757.index.deleted (kafka.log.OffsetIndex)
Deleting segment 0 from log __consumer_offsets-14. (kafka.log.Log)
Deleting segment 1 from log __consumer_offsets-14. (kafka.log.Log)
Deleting index /data/kafka-logs/__consumer_offsets-14/00000000000000000001.index.deleted (kafka.log.OffsetIndex)
Deleting index /data/kafka-logs/__consumer_offsets-14/00000000000000000000.index.deleted (kafka.log.OffsetIndex)
[GroupCoordinator 1001]: Preparing to restabilize group GROUP_NAME with old generation 37 (kafka.coordinator.GroupCoordinator)
[GroupCoordinator 1001]: Stabilized group GROUP_NAME generation 38 (kafka.coordinator.GroupCoordinator)
[GroupCoordinator 1001]: Assignment received from leader for group GROUP_NAME for generation 38 (kafka.coordinator.GroupCoordinator)
[Group Metadata Manager on Broker 1001]: Removed 8 expired offsets in 1 milliseconds. (kafka.coordinator.GroupMetadataManager)
事实上，我有两个问题:
In fact, I have 2 questions:
对于消费者组来说，这个抵消过期是如何工作的?
How does this offset expiration work for a consumer group?
这个过期的偏移量能否解释这种行为，即我的消费者在具有 auto.offset.reset = latest 时不会轮询任何内容，但是当它具有 auto.offset.reset = latest 时它从上次提交的偏移量开始轮询代码>auto.offset.reset = 最早 ?
Can this expired offset explain this behaviour where my consumer would not poll anything when it had auto.offset.reset = latest, but it polled from the last committed offset when it had auto.offset.reset = earliest ?
推荐答案
更新
从 Apache Kafka 2.1 开始，只要消费者组处于活动状态，偏移量就不会被删除，无论消费者是否提交偏移量都是独立的，即 offset.retention.minutes 时钟只开始当组变空时打勾(在较早的版本中，当提交发生时时钟直接开始打勾).
Since Apache Kafka 2.1, offsets won't be deleted as long as the consumer group is active, independent if the consumers commit offsets or not, ie, the offset.retention.minutes clocks only starts to tick when the group becomes empty (in older released, the clock started to tick directly when the commit happened).
原答案
Kafka，默认情况下会在可配置的时间段后删除已提交的偏移量.参见参数offsets.retention.minutes.即，如果消费者组在这段时间内处于非活动状态(即未提交任何偏移量)，则偏移量将被删除.因此，即使消费者正在运行，如果它没有为某些分区提交偏移量，这些偏移量也会受到 offset.retention.minutes 的约束.
Kafka, by default deletes committed offsets after a configurable period of time. See parameter offsets.retention.minutes. Ie, if a consumer group is inactive (ie, does not commit any offsets) for this amount of time, the offsets get deleted. Thus, even if the consumer is running, if it does not commit offsets for some partitions, those offsets are subject to offset.retention.minutes.
如果您启动消费者，则会发生以下情况:
If you start a consumer, the following happens:
- 寻找一个(有效的)已提交的偏移量(对于消费者组)
- look for a (valid) committed offset (for the consumer group)
- 如果找到有效的偏移量，则从那里恢复
- 如果没有找到有效的偏移量，根据
auto.offset.reset参数重置偏移量
- if valid offset is found, resume from there
- if no valid offset is found, reset offset according to
auto.offset.resetparameter
因此，如果您的偏移量被删除并且 auto.offset.reset = latest，您的消费者将不会轮询任何内容，直到将新数据添加到主题中.如果 auto.offset.reset = early 它应该消耗整个主题.
Thus, if your offsets got deleted and auto.offset.reset = latest, you consumer will not poll anything until new data is added to the topic. If auto.offset.reset = earliest it should consume the whole topic.
有关此问题的讨论，请参阅此 JIRA https://issues.apache.org/jira/browse/KAFKA-3806 和 https://issues.apache.org/jira/browse/KAFKA-4682
See this JIRA for a discussion about this https://issues.apache.org/jira/browse/KAFKA-3806 and https://issues.apache.org/jira/browse/KAFKA-4682
这篇关于Apache Kafka 消费者组的偏移量如何到期?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
