如何在Spring-Kafka中新增读最新偏移量的消费者组且不影响其他组

完全可以在处理器级别为新增的消费者组单独覆盖auto.offset.reset参数为latest，且不会对其他已存在的消费者组产生任何影响。

核心原理 #

Kafka的每个消费者组都独立维护自己的偏移量记录，且Spring-Kafka支持在具体消费者/处理器层级覆盖全局配置，不同消费者组的配置相互隔离。

具体实现方式 #

方式一：通过@KafkaListener注解直接指定属性

在新增消费者组的监听方法上，通过properties参数直接覆盖配置：

@KafkaListener(
    topics = "your-target-topic",
    groupId = "new-consumer-group",
    properties = {
        ConsumerConfig.AUTO_OFFSET_RESET_CONFIG + "=latest"
    }
)
public void processNewGroupMessages(ConsumerRecord<String, String> record) {
    // 消息处理逻辑
}

此配置仅对new-consumer-group生效，其他使用全局配置的消费者组仍会采用earliest策略。

方式二：自定义ListenerContainerFactory

如果需要复用该配置，可以单独创建一个容器工厂，覆盖全局配置：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> latestOffsetContainerFactory(ConsumerFactory<String, String> baseConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(baseConsumerFactory);

    // 覆盖auto.offset.reset配置
    Map<String, Object> overrideConfigs = new HashMap<>();
    overrideConfigs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
    factory.getContainerProperties().setKafkaConsumerProperties(overrideConfigs);

    return factory;
}

然后在监听方法中指定该工厂：

@KafkaListener(
    topics = "your-target-topic",
    groupId = "new-consumer-group",
    containerFactory = "latestOffsetContainerFactory"
)
public void processNewGroupMessages(ConsumerRecord<String, String> record) {
    // 消息处理逻辑
}

注意事项 #

由于是新增的消费者组，Kafka集群中没有该组的偏移量记录，auto.offset.reset参数才会生效。如果该组后续产生了偏移量记录，此参数将不再起作用（如需重新调整，需手动重置偏移量）。

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