Spring Boot @KafkaListener消费多topic时如何在catch块获取异常对应topic名

你遇到@Header注入报错的核心原因是：当前你使用的是批量消费模式，@KafkaListener 接收的是整批消息的List<ConsumerRecord>，此时@Header(KafkaHeaders.RECEIVED_TOPIC)如果声明为单个String类型，框架无法将整批多条消息对应的多个topic值映射到单个字符串，因此会抛出类型不匹配错误。

方案1：调整异常捕获粒度（更推荐） #

你本身已经在循环中遍历每条ConsumerRecord，只需要将try-catch的范围缩小到单条消息的处理逻辑，就可以直接从当前处理的ConsumerRecord中获取topic，修改后代码如下：

@KafkaListener(topics = "#{'${app.kafka.consumer.topic}'.split(',')}",
        containerFactory = "kafkaListenerContainerFactory",
        groupId = "${app.kafka.consumer.group-id}")
public void receivedMessage(@Payload List<ConsumerRecord<String, String>> records,
                            Acknowledgment acknowledgment) {
    for (ConsumerRecord<String, String> record : records){
        try{
            process(record.value(),record.topic(), acknowledgment);
        }catch(Exception ex){
            // 直接从当前record获取topic
            log.error("消费消息异常，topic: {}, 偏移量: {}, 异常信息: ", record.topic(), record.offset(), ex);
        }
    }
    // 整批处理完成再提交，或者根据你的业务需求调整提交逻辑
    acknowledge(acknowledgment);
}

这种方案的优势是逻辑简单直观，还可以针对单条消息消费失败做定制化处理，比如跳过失败消息、单独重试等，不会因为单条消息异常影响整批其他消息的消费。

方案2：通过注入批量Header获取 #

如果你不想调整try-catch的范围，可以将注入的topic header声明为List<String>类型，顺序和records列表一一对应，代码如下：

@KafkaListener(topics = "#{'${app.kafka.consumer.topic}'.split(',')}",
        containerFactory = "kafkaListenerContainerFactory",
        groupId = "${app.kafka.consumer.group-id}")
public void receivedMessage(@Payload List<ConsumerRecord<String, String>> records,
                            @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,
                            Acknowledgment acknowledgment) {
    int currentIndex = 0;
    try{
        for (ConsumerRecord<String, String> record : records){
            process(record.value(),record.topic(), acknowledgment);
            currentIndex++;
        }
        acknowledge(acknowledgment);
    }catch(Exception ex){
        // 根据当前处理到的索引获取对应topic
        log.error("消费消息异常，topic: {}, 异常信息: ", topics.get(currentIndex), ex);
    }
}

注意这种方案整批中只要有一条消息异常，后续的消息都会终止处理，你需要根据业务的容错要求选择合适的方案。

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