Spring-Kafka：如何将RecordHeaders以可读格式打印？

当然可以将RecordHeader中byte[]类型的value转换为人类可读的字符串格式打印，以下是几种实用的实现方案：

1. 消费时手动转换并打印 #

在处理ConsumerRecord的逻辑中，直接遍历headers集合，将每个RecordHeader的value通过指定编码（通常为UTF-8）转换为字符串：

import org.apache.kafka.common.header.Header;
import java.nio.charset.StandardCharsets;

// 假设consumerRecord是获取到的消息实例
for (Header header : consumerRecord.headers()) {
    String key = header.key();
    String valueStr = new String(header.value(), StandardCharsets.UTF_8);
    System.out.printf("Header: key=%s, value=%s%n", key, valueStr);
}

注意：必须确保生产消息时使用的编码和此处一致，否则会出现乱码。

2. 重写RecordHeader的toString方法 #

如果你的业务场景需要频繁打印headers，可以自定义一个继承自RecordHeader的类，重写toString方法实现自动转换：

import org.apache.kafka.common.header.internals.RecordHeader;
import java.nio.charset.StandardCharsets;

public class ReadableRecordHeader extends RecordHeader {
    public ReadableRecordHeader(String key, byte[] value) {
        super(key, value);
    }

    @Override
    public String toString() {
        String valueStr = getValue() != null ? new String(getValue(), StandardCharsets.UTF_8) : "null";
        return String.format("RecordHeader(key=%s, value=%s)", getKey(), valueStr);
    }
}

在生产消息时使用这个自定义类，或者消费时将原有RecordHeader转换为该类，打印时即可直接输出可读字符串。

3. 配置日志框架自定义转换器 #

若使用Logback等日志框架，可以自定义一个转换器来处理RecordHeaders的格式化：

<!-- Logback配置示例 -->
<conversionRule conversionWord="kafkaHeaders" converterClass="com.yourpackage.KafkaHeadersConverter" />

对应的转换器类实现：

import ch.qos.logback.classic.pattern.ClassicConverter;
import ch.qos.logback.classic.spi.ILoggingEvent;
import org.apache.kafka.common.header.Headers;
import java.nio.charset.StandardCharsets;

public class KafkaHeadersConverter extends ClassicConverter {
    @Override
    public String convert(ILoggingEvent event) {
        Object[] args = event.getArgumentArray();
        for (Object arg : args) {
            if (arg instanceof Headers) {
                StringBuilder sb = new StringBuilder("RecordHeaders(headers=[");
                for (Header header : ((Headers) arg)) {
                    String valueStr = new String(header.value(), StandardCharsets.UTF_8);
                    sb.append(String.format("RecordHeader(key=%s, value=%s), ", header.key(), valueStr));
                }
                if (sb.length() > 20) {
                    sb.setLength(sb.length() - 2);
                }
                sb.append("])");
                return sb.toString();
            }
        }
        return event.getMessage();
    }
}

之后在日志模板中使用%kafkaHeaders即可自动格式化headers输出。

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