Spring Boot Cron任务修改周期后Kafka Consumer重复消费已处理记录问题

首先，我得先拆解一下你遇到的问题：短间隔（15分钟）的Cron任务消费正常，但改成2小时后就开始重复消费大量旧记录，哪怕只发了3条消息也能收到几千条重复的。结合你的代码和Kafka的工作机制，我梳理了几个核心原因和对应的解决方案：

核心问题分析 #

你的代码里每次Cron任务执行都会新建一个KafkaConsumer实例，用完就close。这种短生命周期的Consumer在间隔较长时，很容易触发Kafka集群的偏移量清理或者消费者组状态异常：

1. 偏移量未被正确持久化：当Consumer关闭后，消费者组会进入 inactive 状态，如果 inactive 时间超过集群的group.retention.ms（默认24小时）或者偏移量 retention 时间（默认7天），集群会删除该组的偏移量记录。下次新Consumer加入时，因为找不到已提交的偏移量，会按照auto.offset.reset规则重置——但你设置的是latest，理论上应该从最新位置开始，除非这个配置没生效或者偏移量被意外清理。
2. 无参数commitSync的潜在风险：你用的commitSync()会提交当前Consumer的"position"（也就是最后一次poll到的记录的offset+1），但如果poll过程中出现异常，或者集群网络波动，可能导致提交失败，而你的代码只是打印异常没有处理，下次任务就会从之前的偏移量重新消费。
3. 短生命周期Consumer的开销：每次创建Consumer都要和集群协调器握手、拉取偏移量，这个过程很容易出现状态不一致，尤其是间隔较长时，协调器可能已经把之前的消费者组标记为已过期。

具体解决方案 #

方案1：改用Spring Kafka的长生命周期Consumer（推荐） #

Spring Kafka的@KafkaListener会帮你管理Consumer的生命周期，避免每次任务都新建/关闭Consumer的问题。具体步骤：

1. 配置消费者工厂和监听容器

@Configuration
public class KafkaConfig {
    @Bean
    public ConsumerFactory<String, CostKafkaModel> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10000");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "cronKafkaConsumer");
        return new DefaultKafkaConsumerFactory<>(props, 
                new StringDeserializer(), 
                new JsonDeserializer<>(CostKafkaModel.class));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, CostKafkaModel> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, CostKafkaModel> factory = 
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        // 手动提交偏移量
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return factory;
    }
}

2. 编写监听器和Cron任务

@Component
public class CostKafkaProcessor {
    // 用线程安全的集合缓存消费到的记录
    private final List<CostKafkaModel> cachedRecords = new CopyOnWriteArrayList<>();

    @KafkaListener(topics = "deltaCosts", groupId = "cronKafkaConsumer")
    public void consume(ConsumerRecord<String, CostKafkaModel> record, Acknowledgment ack) {
        cachedRecords.add(record.value());
        // 手动提交当前记录的偏移量
        ack.acknowledge();
    }

    @Scheduled(cron = "0 0 */2 * * ?") // 每2小时执行一次
    public void compareWithSql() {
        // 把缓存的记录复制出来，避免处理过程中新增记录干扰
        List<CostKafkaModel> recordsToProcess = new ArrayList<>(cachedRecords);
        cachedRecords.clear();
        
        // 这里写你的SQL对比逻辑
        System.out.println("Processing " + recordsToProcess.size() + " records from Kafka");
    }
}

这种方式下，Consumer会一直保持活跃，偏移量能稳定提交，Cron只负责定时处理缓存的记录，完全避免了重复消费的问题。

方案2：修复手动管理Consumer的代码 #

如果必须保持原有的Cron+手动创建Consumer的方式，你需要优化偏移量提交逻辑，并确认偏移量被正确持久化：

1. 手动指定提交的偏移量：不要依赖无参数的commitSync()，明确提交每个分区的最后一条记录的offset+1，确保提交位置准确：

public List<CostKafkaModel> consumeCosts() {
    KafkaConsumer<String, CostKafkaModel> consumer = new KafkaConsumer<>(
            getKafkaConsumerProps(),
            new StringDeserializer(),
            new JsonDeserializer<>(CostKafkaModel.class));
    List<CostKafkaModel> kafkaModelList = new ArrayList<>();
    try {
        consumer.subscribe(Arrays.asList("deltaCosts"));
        ConsumerRecords<String, CostKafkaModel> records = consumer.poll(Duration.ofMillis(1000));
        
        // 按分区处理并提交偏移量
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, CostKafkaModel>> partitionRecords = records.records(partition);
            for (ConsumerRecord<String, CostKafkaModel> record : partitionRecords) {
                kafkaModelList.add(record.value());
            }
            // 获取当前分区最后一条记录的offset+1
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset() + 1;
            // 手动提交该分区的偏移量
            consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset)));
            System.out.println("Committed offset " + lastOffset + " for partition " + partition.partition());
        }
    } catch (CommitFailedException e) {
        System.err.println("Failed to commit offsets: " + e.getMessage());
        // 可以在这里添加重试逻辑
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        consumer.close();
    }
    return kafkaModelList;
}

2. 验证偏移量提交状态：用Kafka的命令行工具检查消费者组的偏移量是否正确提交：

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group cronKafkaConsumer

如果输出里的CURRENT-OFFSET等于LOG-END-OFFSET，说明偏移量提交成功；如果CURRENT-OFFSET一直停留在0或者旧值，说明提交失败，需要排查集群权限、网络或者__consumer_offsets主题的状态。

3. 调整集群配置：如果你的Kafka集群修改了默认的偏移量保留时间，确保offsets.retention.minutes大于2小时，group.retention.ms大于任务间隔，避免偏移量或消费者组被意外清理。

额外注意事项 #

• 不要在Cron任务中重复创建Consumer，这种方式不仅容易出问题，还会给集群带来不必要的开销。
• 确保group.id唯一，不要和其他服务的消费者组重复，避免偏移量被意外覆盖。
• 生产环境中，一定要给偏移量添加重试逻辑，避免因网络波动导致提交失败。

内容的提问来源于stack exchange，提问作者F. Aamir