Spring Kafka批量消费：如何直接将整批消息发至DLT且不重试

针对Spring Kafka 2.8.6版本，我们可以通过自定义异常+错误处理器配置的方式，精准匹配你提出的三种业务场景，核心思路是用专属异常标记整批失败场景，让错误处理器直接触发DLT流程且跳过重试。

具体实现步骤 #

1. 定义整批失败的自定义异常 #

创建一个异常类，用来标记「整批消息需要直接发送DLT且不重试」的场景：

public class WholeBatchFailureException extends RuntimeException {
    public WholeBatchFailureException(String message) {
        super(message);
    }
}

2. 调整错误处理器配置 #

修改commonErrorHandler Bean，将自定义异常加入不重试异常列表，确保触发该异常时直接走DLT流程：

@Bean
public CommonErrorHandler commonErrorHandler(KafkaTemplate<String, Object> kafkaTemplate) {

    ExponentialBackOffWithMaxRetries exponentialBackOffWithMaxRetries = new ExponentialBackOffWithMaxRetries(5);
    exponentialBackOffWithMaxRetries.setInitialInterval(myVal); // 替换为你的配置值
    exponentialBackOffWithMaxRetries.setMultiplier(myVal);     // 替换为你的配置值
    exponentialBackOffWithMaxRetries.setMaxInterval(myVal);     // 替换为你的配置值

    DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate,
            (record, exception) -> new TopicPartition(record.topic() + "-dlt", record.partition()));

    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoverer, exponentialBackOffWithMaxRetries);
    // 添加所有不需要重试的异常，包括自定义的整批失败异常
    errorHandler.addNotRetryableExceptions(ParseException.class, EventHubNonRetryableException.class, WholeBatchFailureException.class);
    return errorHandler;
}

3. 监听器中按场景抛出对应异常 #

在批量监听器方法中，根据不同业务场景抛出对应异常：

@KafkaListener(topics = "your-topic", containerFactory = "kafkaListenerContainerFactory")
public void listen(List<ConsumerRecord<String, Object>> records) {
    try {
        // 调用API的业务逻辑
        boolean apiAvailable = callYourApi(records);
        
        if (!apiAvailable) {
            // 场景3：API不可用，整批直接发DLT，抛自定义异常
            throw new WholeBatchFailureException("API不可用，整批消息转DLT");
        }

        // 模拟场景2：部分消息处理失败
        List<Integer> failedIndices = new ArrayList<>();
        for (int i = 0; i < records.size(); i++) {
            if (isRecordInvalid(records.get(i))) {
                failedIndices.add(i);
            }
        }
        if (!failedIndices.isEmpty()) {
            // 场景2：指定失败消息索引，直接发DLT不重试
            throw new BatchListenerFailedException(failedIndices, new ParseException("部分消息格式错误", 0));
        }

    } catch (SomeRetryableException e) {
        // 场景1：需要重试的异常，直接抛出，错误处理器会重试5次后转DLT
        throw e;
    }
}

各场景实现说明 #

• 场景1（重试后发DLT）：抛出未加入notRetryableExceptions的异常（如自定义的SomeRetryableException），错误处理器会按配置的指数退避策略重试5次，重试失败后将每条消息发送至DLT。
• 场景2（部分消息直接发DLT）：抛出BatchListenerFailedException并指定失败消息的索引，错误处理器仅将指定索引的消息发送至DLT，其余消息视为处理成功，提交偏移量。
• 场景3（整批直接发DLT）：抛出WholeBatchFailureException，由于该异常已被标记为不重试，错误处理器会直接将整批所有消息发送至DLT，随后提交偏移量，不会触发重试。

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