如何在Spring Kafka监听器中实现批处理无限次重试？

针对你的需求，有两种可行方案实现批处理的无限次重试，无需依赖DLT且不需要重启应用即可触发重试：

方案一：利用SeekToCurrentBatchErrorHandler自动重试（推荐） #

Spring Kafka提供的SeekToCurrentBatchErrorHandler可以在批处理失败时自动将consumer的offset回退到当前批次的起始位置，实现重复拉取重试。只要不配置死信队列（DLT）的恢复器，就能避免消息被转发到DLT，实现无限次重试。

步骤1：配置自定义错误处理器 #

创建SeekToCurrentBatchErrorHandler实例，设置重试间隔和无限次重试次数：

@Bean
public SeekToCurrentBatchErrorHandler batchErrorHandler() {
    // 设置重试间隔为5秒，无限次重试（FixedBackOff.UNLIMITED_ATTEMPTS）
    FixedBackOff backOff = new FixedBackOff(5000L, FixedBackOff.UNLIMITED_ATTEMPTS);
    SeekToCurrentBatchErrorHandler errorHandler = new SeekToCurrentBatchErrorHandler();
    errorHandler.setBackOff(backOff);
    // 不要配置DeadLetterPublishingRecoverer，避免消息进入DLT
    return errorHandler;
}

步骤2：配置批处理监听器容器工厂 #

在容器工厂中启用批处理、设置手动确认模式，并绑定自定义错误处理器：

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> batchKafkaListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory,
        SeekToCurrentBatchErrorHandler batchErrorHandler) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setBatchListener(true); // 启用批处理模式
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.setBatchErrorHandler(batchErrorHandler); // 绑定错误处理器
    return factory;
}

步骤3：编写监听器方法 #

处理消息时，若需要重试直接抛出异常，错误处理器会自动触发offset回退和重试：

@KafkaListener(topics = "your-topic-name", containerFactory = "batchKafkaListenerContainerFactory")
public void listenBatch(List<String> messages, Acknowledgment acknowledgment) {
    try {
        // 批量处理消息逻辑
        processBatch(messages);
        // 处理成功后手动确认offset
        acknowledgment.acknowledge();
    } catch (Exception e) {
        // 抛出异常触发错误处理器的重试逻辑
        throw new RuntimeException("Batch processing failed, trigger retry", e);
    }
}

方案二：手动控制Offset回退与重试 #

如果需要更灵活的重试逻辑（比如根据自定义条件决定是否重试），可以手动操作consumer的offset，将其回退到当前批次的起始位置，实现重复拉取。

监听器方法实现 #

注入Consumer和ConsumerRecords对象，处理失败时手动回退offset并添加重试延迟：

@KafkaListener(topics = "your-topic-name", containerFactory = "batchKafkaListenerContainerFactory")
public void listenBatch(
        @Header(KafkaHeaders.RECORDS) ConsumerRecords<String, String> records,
        Acknowledgment acknowledgment,
        @Header(KafkaHeaders.CONSUMER) Consumer<String, String> consumer) {
    boolean needRetry = false;
    try {
        // 批量处理消息逻辑
        processBatch(records);
        acknowledgment.acknowledge();
    } catch (Exception e) {
        needRetry = true;
    }

    if (needRetry) {
        // 遍历当前批次的所有分区，将offset回退到批次起始位置
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
            long startOffset = partitionRecords.get(0).offset();
            consumer.seek(partition, startOffset);
        }

        // 添加重试延迟，避免频繁重试占用资源
        try {
            Thread.sleep(5000); // 延迟5秒后重试
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
        }
        // 不调用acknowledge()，保留当前offset未提交状态
    }
}

注意事项 #

• 两种方案均需确保consumer的enable.auto.commit配置为false（Spring Kafka默认在手动确认模式下会自动设置为false）
• 重试间隔需根据业务场景合理设置，避免给Kafka集群带来过大压力
• 方案一中，若后续需要添加DLT逻辑，只需为SeekToCurrentBatchErrorHandler配置DeadLetterPublishingRecoverer即可

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