Kafka Broker偏移量/日志保留和最早模式下的使用者偏移量重置

Question

问题描述：

我们的Kafka consumer（在Spring Boot2.x中开发）正在执行几天。当我们重新启动这些消费者时，主题的所有消息都将被再次消费，但仅在特定条件下。

条件：

           auto.commit.interval.ms = 100
           auto.offset.reset = earliest
           bootstrap.servers = [server1:9092]
           check.crcs = true
           client.id = 
           connections.max.idle.ms = 540000
           enable.auto.commit = false
           exclude.internal.topics = true
           fetch.max.bytes = 52428800
           fetch.max.wait.ms = 500
           fetch.min.bytes = 1
           group.id = consumer_group1
           heartbeat.interval.ms = 3000
           interceptor.classes = null
           internal.leave.group.on.close = true
           isolation.level = read_uncommitted
           key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
           max.partition.fetch.bytes = 1048576
           max.poll.interval.ms = 300000
           max.poll.records = 500
           metadata.max.age.ms = 300000
           metrics.recording.level = INFO
           metrics.sample.window.ms = 30000
           partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
           receive.buffer.bytes = 65536
           reconnect.backoff.max.ms = 1000
           reconnect.backoff.ms = 50
           request.timeout.ms = 305000
           retry.backoff.ms = 100
           value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer

代理配置：

           log.retention.ms = 86400000
           log.retention.minutes = 10080
           log.retention.hours = 168
           log.retention.bytes = -1

           offsets.retention.ms = 864000000
           offsets.retention.minutes = 14400
           offsets.retention.hours = 240 

           unclean.leader.election.enable = false
           log.cleaner.enable = true
           auto.leader.rebalance.enable = true
           leader.imbalance.check.interval.seconds = 300
           log.retention.check.interval.ms = 300000
           log.cleaner.delete.retention.ms = 604800000

谢谢和问候

Answer 1

您是对的，您遇到此问题的原因是log.retention.*和 ；offsets.retention.*的值不同（分别为7天和1天）。对于2.0之前的Kafka版本，请检查此处的说明。这是由于很少的消息进入您的主题，并且偏移数据已经过期。

关于您的短语这是不完全正确的，显然我们不能将consumer设置为 ；“latest”。如果您收到上次邮件的时间少于1天（如几小时前），则可以安全地将auto.offset.reset值更新为最新，并使用相同的组id（或application.id)。在这种情况下，您不会丢失消息。

作为另一种选择，您可以将特定主题的日志保留值更改为1天。此外，您还可以更新 ；valueoffsets.retention.*，但您需要从您的性能点测试它，它可能会被降级。