Spring Kafka批量消费时，如何避免异常重试已确认的消息？

问题场景 #

使用Spring Kafka自动配置的默认消费者/监听器工厂，拉取1000条记录批量处理，监听器代码如下：

@KafkaListener(topics = "abcd-topic")
public void processRecords(List<ConsumerRecord<String, Word>> consumerRecords)

Spring会一次性传入整批1000条记录，当监听器处理期间抛出异常时，默认会重试整个批次。即使设置了spring.kafka.listener.ack-mode=manual_immediate并中途确认部分记录，异常时仍会重试全量（推测容器在内存中保留了批次记录），且使用DefaultErrorHandler处理异常。

解决方案 #

1. 修正手动确认的正确姿势 #

批量监听器中，无法直接通过acknowledge(counter)提交部分记录的offset（标准Acknowledgment接口无此方法），需通过消费者实例手动提交指定分区的offset：

@KafkaListener(topics = "abcd-topic")
public void processRecords(List<ConsumerRecord<String, Word>> consumerRecords, Consumer<String, Word> consumer) {
    int counter = 0;
    final int OFFSET_COMMIT_BATCH_SIZE = 100;
    Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

    for (ConsumerRecord<String, Word> record : consumerRecords) {
        // 单条记录处理逻辑
        processSingleRecord(record);
        counter++;

        if (counter % OFFSET_COMMIT_BATCH_SIZE == 0) {
            log.info("Processed {} records", counter);
            // 提交当前分区的offset（当前记录offset + 1，表示已处理完该记录）
            TopicPartition tp = new TopicPartition(record.topic(), record.partition());
            offsetsToCommit.put(tp, new OffsetAndMetadata(record.offset() + 1));
            consumer.commitSync(offsetsToCommit);
            offsetsToCommit.clear();
        }
    }

    // 提交剩余未批量提交的记录
    if (!offsetsToCommit.isEmpty()) {
        consumer.commitSync(offsetsToCommit);
    }
}

2. 配置错误处理器，过滤已确认记录 #

默认DefaultErrorHandler会重试整个批次，需自定义批量重试策略，仅处理未提交offset的记录：

@Bean
public DefaultErrorHandler batchErrorHandler() {
    // 定义异常最终处理逻辑（如死信队列）
    Consumer<ConsumerRecord<?, ?>, Exception> recoverer = (record, ex) -> {
        log.error("Failed to process record after retries, sending to DLQ", ex);
        // 此处可添加死信队列发送逻辑
    };

    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));
    // 设置批量重试策略，过滤已提交offset的记录
    errorHandler.setBatchRetryStrategy((context, records) -> {
        // 需维护已提交的offset集合，这里简化为仅保留当前批次中未处理完成的记录
        return records.stream()
                .filter(record -> !isOffsetCommitted(record))
                .collect(Collectors.toList());
    });
    return errorHandler;
}

// 辅助方法：判断记录的offset是否已提交
private boolean isOffsetCommitted(ConsumerRecord<?, ?> record) {
    TopicPartition tp = new TopicPartition(record.topic(), record.partition());
    OffsetAndMetadata committed = consumer.committed(tp);
    return committed != null && committed.offset() > record.offset();
}

3. 调整容器配置，禁用全批次内存重试 #

若不需要内存中保留批次重试，可直接限制重试次数或禁用重试，让异常记录进入死信队列：

# 禁用重试，直接进入死信队列
spring.kafka.listener.max-retries=0
# 死信队列前缀，自动生成死信队列（如dlq-abcd-topic）
spring.kafka.listener.dead-letter-topic-prefix=dlq-

核心逻辑 #

确保已处理完成的记录offset被正确提交至Broker，同时通过错误处理器过滤掉已提交offset的记录，避免重复重试。

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