Kafka批量消费者Offset提交处理：Spring-Kafka 2.2.8场景咨询

当然可以实现这种需求！在Spring Kafka 2.2.8里，你需要跳出默认的批量全提交逻辑，通过手动控制偏移量和拆分处理来实现「成功记录提交偏移量，失败记录转存」的效果，具体可以按照下面的步骤来做：

1. 调整消费者配置，开启手动提交 #

首先要关闭自动提交偏移量，让我们能自主控制提交时机：

spring.kafka.consumer.enable-auto-commit=false
spring.kafka.listener.ack-mode=MANUAL_IMMEDIATE # 或MANUAL，按需选择

配置后，消费者不会自动提交偏移量，需要我们在代码里手动触发提交操作。

2. 拆分批量记录，逐个处理并捕获异常 #

不要直接对整个批量做统一处理，而是遍历每一条记录单独执行业务逻辑，专门捕获瞬时异常（比如数据库短暂宕机、网络波动这类可恢复的错误）。

举个代码示例：

@KafkaListener(topics = "your-source-topic", containerFactory = "batchListenerContainerFactory")
public void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
    List<ConsumerRecord<String, String>> failedRecords = new ArrayList<>();

    for (ConsumerRecord<String, String> record : records) {
        try {
            // 执行你的业务处理逻辑
            processRecord(record);
        } catch (TransientDatabaseException | NetworkException e) { // 捕获你的瞬时错误类型
            // 暂存失败记录
            failedRecords.add(record);
            log.error("处理记录失败，offset: {}, value: {}", record.offset(), record.value(), e);
        }
    }

    // 先处理失败记录的转存
    if (!failedRecords.isEmpty()) {
        sendToDeadLetterTopic(failedRecords);
        // 也可以选择持久化到数据库：saveFailedRecordsToDB(failedRecords);
    }

    // 提交整个批量的偏移量（失败记录已转存，无需重复消费）
    ack.acknowledge();
}

3. 实现失败记录的转存逻辑 #

对于失败记录，你可以选择发送到死信主题（DLT），或者持久化到数据库/文件系统。Spring Kafka 2.2.8已经提供了DeadLetterPublishingRecoverer来简化死信主题的发送，示例如下：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

private void sendToDeadLetterTopic(List<ConsumerRecord<String, String>> failedRecords) {
    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, ex) -> new TopicPartition("your-dlt-topic", record.partition()));
    
    for (ConsumerRecord<String, String> failedRecord : failedRecords) {
        recoverer.accept(failedRecord, new RuntimeException("瞬时处理失败"));
    }
}

后续可以专门启动一个消费者监听死信主题，等下游系统恢复后重新处理这些记录。

4. 关键注意事项 #

• 可靠性优先：一定要确保失败记录成功转存后再提交偏移量，避免提交后转存失败导致记录丢失。如果转存过程也可能出错，可以先将失败记录写入本地磁盘，再异步转存。
• 异常范围控制：只捕获可恢复的瞬时异常，对于数据格式错误这类不可恢复的异常，也可以按同样逻辑转存，后续人工介入处理。
• 批量性能：拆分逐个处理会增加一定的开销，但对于瞬时错误占比低的场景，这种开销完全可接受。

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