如何在Spring Kafka的AbstractConsumerSeekAware类中正确实现seekToEnd()？

问题根因 #

你现有代码不生效的核心原因是seekToEnd()的执行时机错误：

• 该方法写在@KafkaListener注解的消费方法中，执行时当前批次的消息已经通过poll操作拉取到本地，seek操作只会对后续的拉取请求生效
• 应用重启后第一次拉取消息会优先读取Broker上存储的消费者组已提交偏移量，此时还未触发消费方法内的seek逻辑，所以会继续从上次停止的位置消费

正确实现方式 #

方案1：重写AbstractConsumerSeekAware的分区分配回调（推荐） #

onPartitionsAssigned方法会在消费者完成分区分配、第一次拉取消息之前触发，在此处执行seek操作可以确保首次拉取就从最新偏移量开始，代码示例如下：

public class KafkaConsumer extends AbstractConsumerSeekAware {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // 分区分配完成后回调，早于首次消息拉取
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        // 将当前消费者分配到的所有分区都seek到末尾
        callback.seekToEnd(assignments.keySet());
    }

    @KafkaListener(topics = "${topic.consumer}")
    public void consume(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
        // 执行业务逻辑
        doSomething();
        ack.acknowledge();
    }

    private void doSomething() {
        // 你的业务处理逻辑
    }
}

方案2：动态生成消费组ID（临时场景适用） #

如果是测试等临时场景，不需要保留消费组的历史记录，可以每次启动应用时生成随机的group.id，配合Kafka消费者配置auto.offset.reset=latest，也能实现从最新偏移量消费，缺点是会在Kafka集群产生大量废弃的消费组元数据，生产环境不推荐长期使用。

注意事项 #

• 多实例部署场景下方案1完全兼容，每个实例只会对自身分配到的分区执行seek操作，不会互相影响
• 不要在消费方法内部频繁执行seek操作，会影响消费性能，还可能导致偏移量错乱

内容的提问来源于stack exchange，提问作者Teja M