Spring Kafka：如何在错误处理器中确认毒丸消息及启用手动Ack？

采用**至少一次（at-least-once）**语义处理Kafka消息，使用Spring Kafka 2.9.5（对应Spring Boot 2.7.8）的异步确认（async ack）功能，正常流程下消息处理完成后能提交偏移量，但遇到毒丸消息（poison pill）时，无法在错误处理器中提交坏记录的偏移量，导致消费者卡在坏消息处无法继续消费新记录。

现有代码如下：

1. Kafka监听器工厂 #

public ConcurrentKafkaListenerContainerFactory<String, JsonNode> kafkaListenerContainerFactory(ConsumerFactory<String, JsonNode> kafkaConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, JsonNode> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setErrorHandler(errorHandler()); // 关键冲突点
        factory.getContainerProperties().setAsyncAcks(true);
        return factory;
    }

2. 原错误处理器（无法获取Acknowledgement） #

@Bean("errorHandler")
public ErrorHandler errorHandler() {
    log.info("Creating error handler");
    return (thrownException, records) -> {
        log.error("Inside error handler");
    };
}

3. 手动确认错误处理器（未被调用） #

@Bean("kafkaListenErrorHandler")
public ManualAckListenerErrorHandler kafkaListenerErrorHandler() {
    return (message, exception, consumer, ack) -> {
        log.info("Inside manual ack error handler " + exception.getMessage());
        exception.printStackTrace();
        ack.acknowledge();
        return null;
    };
}

4. Kafka消费者 #

@KafkaListener(
        id="default_kafka_listener",
        topics = "topic",
        groupId = "groupId",
        containerFactory = "kafkaListener",
        errorHandler = "kafkaListenErrorHandler",
        autoStartup = "false")
public void consume(@Payload JsonNode message)

要让ManualAckListenerErrorHandler生效并能提交错误消息的偏移量，需解决两个核心问题：

1. 移除容器工厂的全局错误处理器配置 #

容器工厂中设置的setErrorHandler(errorHandler())会覆盖@KafkaListener注解上指定的errorHandler（容器级错误处理器优先级更高），直接删除该行配置，让监听器级别的错误处理器生效：

修改后的监听器工厂代码：

public ConcurrentKafkaListenerContainerFactory<String, JsonNode> kafkaListenerContainerFactory(ConsumerFactory<String, JsonNode> kafkaConsumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, JsonNode> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(kafkaConsumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        // 移除factory.setErrorHandler(errorHandler())
        factory.getContainerProperties().setAsyncAcks(true);
        return factory;
    }

2. 在消费者方法中注入Acknowledgement对象 #

手动确认模式下，必须在消费者方法中声明Acknowledgement参数，ManualAckListenerErrorHandler才能获取到对应的确认实例。修改消费者方法：

@KafkaListener(
        id="default_kafka_listener",
        topics = "topic",
        groupId = "groupId",
        containerFactory = "kafkaListener",
        errorHandler = "kafkaListenErrorHandler",
        autoStartup = "false")
public void consume(@Payload JsonNode message, Acknowledgement ack) // 新增Acknowledgement参数

3. 验证错误处理器逻辑 #

你的ManualAckListenerErrorHandler代码逻辑是正确的：调用ack.acknowledge()会提交当前消息的偏移量（开启异步确认后，该操作为异步执行），毒丸消息的偏移量被提交后，消费者即可继续处理后续消息。

• Spring Kafka 2.8+的异步确认功能与ManualAckListenerErrorHandler完全兼容，无需额外配置。
• 若需全局错误处理器，建议使用Spring Kafka 2.8+推荐的CommonErrorHandler，但当前场景下监听器级别的ManualAckListenerErrorHandler更适配特定监听器的错误确认需求。

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