auto-offset-reset=latest 在 spring-kafka 中不起作用
auto-offset-reset=latest does not work in spring-kafka
我有一个用例，我希望消费者始终从最新的偏移量开始。我不需要为这个消费者提交抵消。这是不可能用 spring-kafka 实现的，因为新的消费者组总是提交新分配的分区。然后，在程序的后续启动中，消费者从这个存储的偏移量中读取，而不是从最新的偏移量中读取。换句话说，只有第一次开始使用新的消费者组时行为正确，即从最新的开始消费。问题出在 KafkaMessageListenerContainer$ListenerConsumer.onPartitionsAssigned()
作为参考，我在 spring boot
中设置了以下内容
spring.kafka.listener.ack-mode=manual
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=false
添加该代码是为了解决在新的消费者组开始消费时发生重新分区时出现的一些令人讨厌的竞争条件；它可能会导致记录丢失或重复，具体取决于配置。
最好提交初始偏移量以避免这些情况。
不过，我同意，如果用户对偏移量承担全部责任（使用 MANUAL 确认模式），那么我们可能不应该进行该提交；由用户代码来处理比赛（在您的情况下，您不关心丢失的记录）。
随时打开 GitHub 问题（欢迎投稿）。
同时，您可以通过让您的侦听器实现 ConsumerSeekAware 并在分配期间寻求 topic/partition 结束来避免这种情况。
另一种方法是每次都为 group.id 使用一个 UUID；并且您将始终从主题结尾开始。
我有一个用例，我希望消费者始终从最新的偏移量开始。我不需要为这个消费者提交抵消。这是不可能用 spring-kafka 实现的，因为新的消费者组总是提交新分配的分区。然后，在程序的后续启动中，消费者从这个存储的偏移量中读取，而不是从最新的偏移量中读取。换句话说，只有第一次开始使用新的消费者组时行为正确，即从最新的开始消费。问题出在 KafkaMessageListenerContainer$ListenerConsumer.onPartitionsAssigned()
作为参考，我在 spring boot
中设置了以下内容spring.kafka.listener.ack-mode=manual
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=false
添加该代码是为了解决在新的消费者组开始消费时发生重新分区时出现的一些令人讨厌的竞争条件；它可能会导致记录丢失或重复，具体取决于配置。
最好提交初始偏移量以避免这些情况。
不过，我同意，如果用户对偏移量承担全部责任（使用 MANUAL 确认模式），那么我们可能不应该进行该提交；由用户代码来处理比赛（在您的情况下，您不关心丢失的记录）。
随时打开 GitHub 问题（欢迎投稿）。
同时，您可以通过让您的侦听器实现 ConsumerSeekAware 并在分配期间寻求 topic/partition 结束来避免这种情况。
另一种方法是每次都为 group.id 使用一个 UUID；并且您将始终从主题结尾开始。