如何让重启后的Reactive Kafka消费者跳过未读消息（同消费组）

当前配置 #

配置文件 #

spring.kafka.consumer.group-id=MY-CONSUMER-GROUP

消费者Bean定义 #

@Bean
public ReactiveKafkaConsumerTemplate<UUID, MyEvent> createConsumer(@Autowired KafkaProperties properties) {
    return new ReactiveKafkaConsumerTemplate<>(
        ReceiverOptions.<UUID, MyEvent>create(new HashMap<>(properties.buildConsumerProperties()))
            .consumerProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
            .subscription(List.of("MY-TOPIC")));
}

该配置确保消费者首次创建时仅读取新消息，不会从头消费所有历史消息。

需求说明 #

我们希望服务每次重启后都保持上述行为：仅处理服务运行期间产生的新消息。但Kafka官方文档明确说明：

一旦消费组已写入偏移量，auto.offset.reset配置参数将不再生效。若消费组中的消费者停止后重启，会从最后提交的偏移量处继续消费。

我们的核心目标是：服务停机一段时间后重启，仍使用同一消费组，直接跳过停机期间写入主题的所有消息，从主题末尾（最新消息位置）开始消费。

当前消费者使用方式 #

@EventListener(value = ApplicationReadyEvent.class)
public void listen() {
    consumerTemplate
        .receiveAutoAck()
        .concatMap(consumerRecord -> handleEvent(consumerRecord.key(), consumerRecord.value())
            .onErrorResume(__ -> Mono.empty())
        )
        .subscribe();
}

private Mono<MyEvent> handleEvent(@NonNull UUID key, @NonNull MyEvent event) {
    // 业务处理逻辑
}

疑问 #

我们注意到KafkaConsumer.seekToEnd()方法，但存在两个疑问：

1. 该方法是否能实现我们的需求？
2. 它返回void，如何在Reactive Kafka的响应式调用链中正确使用？

解决方案 #

1. seekToEnd()的作用确认 #

seekToEnd()完全可以满足需求：它会将消费组的偏移量直接定位到指定分区的最新消息位置，后续消费将从这个位置开始，完全跳过停机期间产生的所有消息。

2. 在Reactive Kafka中使用seekToEnd()的正确方式 #

在Reactive Kafka中，我们可以通过ReceiverOptions提供的doOnConsumer钩子来执行seekToEnd()操作。这个钩子会在消费者实例初始化完成、订阅分区之后触发，刚好适合在启动时调整偏移量。

修改后的消费者Bean配置如下：

@Bean
public ReactiveKafkaConsumerTemplate<UUID, MyEvent> createConsumer(@Autowired KafkaProperties properties) {
    ReceiverOptions<UUID, MyEvent> receiverOptions = ReceiverOptions.<UUID, MyEvent>create(new HashMap<>(properties.buildConsumerProperties()))
            .consumerProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest")
            .subscription(List.of("MY-TOPIC"));

    // 添加消费者初始化钩子，执行seekToEnd操作
    receiverOptions = receiverOptions.doOnConsumer(consumer -> {
        // 获取当前消费者订阅的所有分区
        Collection<TopicPartition> assignedPartitions = consumer.assignment();
        // 将所有分区的偏移量定位到末尾
        consumer.seekToEnd(assignedPartitions);
    });

    return new ReactiveKafkaConsumerTemplate<>(receiverOptions);
}

3. 额外注意事项 #

• 该操作会在消费者每次启动时执行，无论消费组是否已有历史偏移量，确保每次重启都从最新消息开始消费。
• 如果需要更灵活的控制（比如仅在特定场景下执行seek操作），可以在doOnConsumer中添加自定义判断逻辑，例如读取配置开关或启动参数。
• doOnConsumer是在消费者线程中同步执行的，不会阻塞响应式链的初始化流程，符合Reactive编程的非阻塞特性。

内容的提问来源于stack exchange，提问作者Honza Zidek