如何在属于 AbstractConsumerSeekAware 类的 Spring Kafka 中实现 seekToEnd()？答案

【问题标题】：how to implement seekToEnd() in Spring Kafka belonging to AbstractConsumerSeekAware class?如何在属于 AbstractConsumerSeekAware 类的 Spring Kafka 中实现 seekToEnd()？
【发布时间】：2021-12-08 13:48:20
【问题描述】：

我有一个要求。我们有一个从 kafka 主题读取的 spring boot kafka 消费者应用程序。我们的要求是每当应用程序出现故障并出现时，我都想从最新的偏移量开始，而不是被旧值所困扰。是否有可能重置组的偏移量？我进行了一些研究，并使用 AbstractConsumerSeekAware 使用 seekToEnd() 将偏移设置到末尾，如下面的代码所示。

public class KafkaConsumer extends AbstractConsumerSeekAware {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;



    @KafkaListener(topics = "${topic.consumer}")
    public void consume(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
        //consuming the message from consumer Record.
        seekToEnd();
        doSomething();
        ack.acknowledge();
    }

但是，当我们停止应用并重新启动它时，它会从上次离开的偏移量开始读取，但我们只想从应用启动时的偏移量读取。我们怎样才能做到这一点？

【问题讨论】：

标签： spring-boot apache-kafka spring-kafka spring-cloud-stream

【解决方案1】：

在@KafkaListener 方法中这样做是不正确的。只有当消费者从分区传递记录时才会真正调用这个。

出于这个原因，您必须实现onPartitionsAssigned()，因此消费者会在开始轮询之前寻找这些分区。

在文档中查看更多信息：https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek

【讨论】：

非常感谢！！你能告诉我这是否好吗？ public class KafkaConsumer implements ConsumerSeekAware{ @Autowired private KafkaTemplate<String, String> kafkaTemplate; @KafkaListener(topics = "${topic.consumer}") public void consume(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) { } @Override public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) { assignments.keySet().forEach(partition -> callback.seekToEnd(assignments.keySet())); } }