Spring Kafka asyncAck模式下异常处理及至少一次消费问题

你遇到的问题很典型：在**手动异步确认（MANUAL + asyncAcks=true）**模式下，正常消息能通过手动提交偏移量正常消费，但遇到格式错误的毒丸消息时，错误处理器里无法提交偏移量，导致消费者一直卡在这条坏消息上。而且旧的ErrorHandler已经过时，我们需要用Spring Kafka推荐的CommonErrorHandler体系来解决。

核心问题分析 #

旧的ErrorHandler不支持与异步确认模式的偏移量提交协同工作，而CommonErrorHandler（尤其是DefaultErrorHandler）是Spring Kafka 2.8+版本针对这类场景设计的替代方案，它能更好地处理偏移量提交、重试和跳过逻辑。

解决方案步骤 #

1. 替换为DefaultErrorHandler（CommonErrorHandler实现）

我们使用DefaultErrorHandler，它允许我们配置恢复策略——比如当消息处理失败时，直接提交该消息的偏移量并跳过，避免重复消费。

2. 配置监听器工厂

修改你的kafkaListenerContainerFactory，将过时的setErrorHandler替换为setCommonErrorHandler，并保留异步确认配置：

public ConcurrentKafkaListenerContainerFactory<String, JsonNode> kafkaListenerContainerFactory(ConsumerFactory<String, JsonNode> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<String, JsonNode> factory =
            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(kafkaConsumerFactory);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
    // 替换为DefaultErrorHandler
    factory.setCommonErrorHandler(defaultErrorHandler());
    factory.getContainerProperties().setAsyncAcks(true);
    return factory;
}

3. 定义DefaultErrorHandler Bean

这里我们配置一个DefaultErrorHandler，设置当处理失败时直接提交偏移量并跳过该消息。你可以根据需求调整重试次数，或者自定义错误处理逻辑：

@Bean
public DefaultErrorHandler defaultErrorHandler() {
    log.info("Creating DefaultErrorHandler for handling poison pills");
    
    // 自定义恢复回调：当消息处理失败时，提交该记录的偏移量
    RecoveryCallback<Void> recoveryCallback = (context) -> {
        ConsumerRecord<?, ?> record = context.getConsumerRecord();
        if (record != null) {
            log.error("处理毒丸消息失败，提交偏移量跳过该消息: topic={}, partition={}, offset={}",
                    record.topic(), record.partition(), record.offset());
            // 提交当前记录的下一个偏移量，跳过坏消息
            context.getConsumer().commitSync(Map.of(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
            ));
        }
        return null;
    };

    // 创建DefaultErrorHandler，设置恢复回调，0次重试直接进入恢复逻辑
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(recoveryCallback, new FixedBackOff(0L, 0));
    
    // 可选：指定哪些异常直接跳过（比如JSON格式错误这类无法重试修复的异常）
    errorHandler.addNotRetryableExceptions(JsonProcessingException.class, IllegalArgumentException.class);
    
    return errorHandler;
}

4. 保留正常消息的手动确认逻辑

在你的Kafka监听器方法中，正常处理完消息后，还是按照之前的逻辑手动提交偏移量即可：

@KafkaListener(topics = "your-topic")
public void listen(ConsumerRecord<String, JsonNode> record, Acknowledgment acknowledgment) {
    try {
        // 处理正常消息逻辑
        processMessage(record.value());
        // 异步确认偏移量
        acknowledgment.acknowledge();
    } catch (Exception e) {
        // 异常会被DefaultErrorHandler捕获，无需手动处理
        throw new RuntimeException("消息处理失败", e);
    }
}

关键说明 #

• FixedBackOff(0L, 0)表示不进行重试，直接进入恢复回调；如果需要重试几次再跳过，可以调整参数（比如FixedBackOff(1000L, 3)表示重试3次，每次间隔1秒）。
• addNotRetryableExceptions可以指定哪些异常不需要重试，比如JSON格式错误这类无法通过重试解决的异常，直接触发跳过逻辑。
• 在恢复回调中，我们使用commitSync提交偏移量（record.offset() + 1表示提交到下一个偏移量），确保消费者能继续消费后续消息。

这样配置后，当遇到毒丸消息时，错误处理器会自动提交该消息的偏移量并跳过，不会阻塞后续消息的消费，同时正常消息的异步确认流程也能正常工作。

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