如何在属于Abstract消费者SeekAware类的Spring Kafka中实现lookToEnd（）？

Question

我有一个要求。我们有一个Spring启动Kafka消费者应用程序，它正在阅读Kafka主题。我们的要求是，每当应用程序出现故障并出现时，我都希望开始最新的偏移量，而不是被旧值所困扰。是否有可能重置组的偏移量？我已经研究了一点点，并使用抽象消费者使用 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();
    }

但是，当我们停止应用程序并重新启动它时，它从它离开的最后一个偏移量开始读取，但我们希望仅在应用程序启动时从偏移量读取。我们怎样才能做到这一点？

Answer 1

在@KafkaListener方法中执行此操作是不正确的。只有当使用者从分区传递记录时，才会真正调用此分区。

出于这个原因，您必须实现onPartitionsAssigned（），因此消费者将在开始轮询之前查找这些分区。

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