Spring-Kafka项目如何记录所有传入的Kafka请求KV键值对？

方案1：修改监听方法接收原始消息 #

直接调整@KafkaHandler方法的入参类型，获取完整的消息元数据和原始内容，无需额外全局配置，适合单个监听器临时排查：

@KafkaHandler(isDefault = true)
@Description(value = "Event received")
public void onEvent(ConsumerRecord<String, String> record) {
    // 打印完整KV和消息属性
    log.info("收到Kafka消息：key={}, value={}, topic={}, partition={}, offset={}",
        record.key(),
        record.value(),
        record.topic(),
        record.partition(),
        record.offset());
    // 手动转换Payload，可捕获转换异常定位问题
    ObjectMapper objectMapper = new ObjectMapper();
    try {
        Payload payload = objectMapper.readValue(record.value(), Payload.class);
        // 后续业务逻辑
    } catch (JsonProcessingException e) {
        log.error("Payload转换失败，原始内容：{}", record.value(), e);
    }
}

方案2：配置全局消息拦截器 #

实现统一的消息打点，无需修改业务监听器代码，适合全链路日志采集：

1. 自定义消息拦截器实现RecordInterceptor接口：

@Component
public class KafkaGlobalLoggingInterceptor implements RecordInterceptor<Object, Object> {
    private static final Logger log = LoggerFactory.getLogger(KafkaGlobalLoggingInterceptor.class);

    @Override
    public ConsumerRecord<Object, Object> intercept(ConsumerRecord<Object, Object> record, Consumer<Object, Object> consumer) {
        // 打印KV、header等所有消息属性
        String headers = StreamSupport.stream(record.headers().spliterator(), false)
            .map(header -> header.key() + "=" + new String(header.value(), StandardCharsets.UTF_8))
            .collect(Collectors.joining(","));
        log.info("Kafka全局接收消息：key={}, value={}, headers={}, topic={}, partition={}, offset={}",
            record.key(), record.value(), headers, record.topic(), record.partition(), record.offset());
        return record;
    }
}

2. Spring Boot项目直接在配置文件中注册拦截器即可生效：

spring.kafka.listener.interceptor-classes=com.yourpackage.KafkaGlobalLoggingInterceptor

方案3：开启消息转换器调试日志 #

针对Payload映射异常的场景，直接调整Jackson消息转换器的日志级别，即可看到完整的转换过程和错误原因，不用改代码：
在配置文件中添加日志级别配置：

logging.level.org.springframework.kafka.support.converter.MappingJackson2MessageConverter=DEBUG
logging.level.com.fasterxml.jackson.databind=DEBUG

Payload字段为null常见排查点 #

• 检查Java类是否提供public无参构造函数
• 检查JSON字段名是否和Java类属性名一致，不一致需添加@JsonProperty注解映射
• 检查字段类型是否匹配，比如JSON字符串无法直接映射到Java Integer字段
• 检查是否配置了Jackson忽略未知属性，多余的未知字段不会导致转换失败，但缺少匹配的字段会为null

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