如何使用Log4j2的KafkaAppender设置Kafka日志消息头

一、给Log4j2 KafkaAppender添加自定义消息头 #

要让Kafka Connect能读取到日志消息的头，首先得确保你的KafkaAppender在发送日志时把需要的元数据塞进Kafka消息头里。Log4j2的KafkaAppender支持通过Property节点配置来添加消息头，示例配置如下：

<Kafka name="KafkaAppender" topic="logs-topic">
  <PatternLayout pattern="%d{ISO8601} [%t] %-5level %logger{36} - %msg%n"/>
  <!-- 将日志级别、Logger名称、线程名作为消息头发送 -->
  <Property name="logLevel">%level</Property>
  <Property name="loggerName">%logger</Property>
  <Property name="threadName">%t</Property>
  <!-- 也可以传递MDC中的自定义字段 -->
  <Property name="businessId">%X{businessId}</Property>
  <ProducerConfig name="bootstrap.servers" value="kafka-broker:9092"/>
</Kafka>

这里用%level、%logger等Log4j2内置占位符，或者MDC中的%X{key}，就能把日志的关键元数据作为Kafka消息Header发送出去。

二、Kafka Connect中基于消息头的谓词判断与转换 #

在Kafka Connect里，你可以通过内置谓词或自定义逻辑，根据消息头的值决定是否应用指定转换。

1. 使用内置谓词匹配消息头 #

Connect的HeaderValueMatches谓词可以直接匹配消息头的值，结合转换实现条件触发。比如只对logLevel头为ERROR的消息添加优先级字段，配置示例如下：

{
  "name": "log-processing-connector",
  "config": {
    "connector.class": "org.apache.kafka.connect.source.FileStreamSourceConnector",
    "tasks.max": "1",
    "topic": "logs-topic",
    "transforms": "filterError,addPriority",
    "transforms.filterError.type": "org.apache.kafka.connect.transforms.Filter$Value",
    "transforms.filterError.predicate": "isErrorLog",
    // 定义谓词：匹配logLevel头为ERROR的消息
    "predicates.isErrorLog.type": "org.apache.kafka.connect.transforms.predicates.HeaderValueMatches",
    "predicates.isErrorLog.header": "logLevel",
    "predicates.isErrorLog.pattern": "^ERROR$",
    // 对匹配上的消息添加优先级字段
    "transforms.addPriority.type": "org.apache.kafka.connect.transforms.InsertField$Value",
    "transforms.addPriority.static.field": "priority",
    "transforms.addPriority.static.value": "high"
  }
}

2. 自定义谓词处理复杂逻辑 #

如果内置谓词满足不了多条件判断等复杂需求，可以自定义Predicate实现类。核心是实现org.apache.kafka.connect.transforms.predicates.Predicate接口，在test方法中读取消息头并执行判断逻辑：

public class MultiHeaderPredicate implements Predicate<RoutedRecord> {
    private String levelHeader;
    private String businessHeader;

    @Override
    public void configure(Map<String, ?> configs) {
        this.levelHeader = (String) configs.get("level-header");
        this.businessHeader = (String) configs.get("business-header");
    }

    @Override
    public boolean test(RoutedRecord record) {
        Headers headers = record.record().headers();
        // 检查日志级别是否为ERROR，且业务ID头存在
        Header levelHeader = headers.lastHeader(this.levelHeader);
        Header businessHeader = headers.lastHeader(this.businessHeader);
        if (levelHeader == null || businessHeader == null) {
            return false;
        }
        String levelValue = new String(levelHeader.value(), StandardCharsets.UTF_8);
        return levelValue.equals("ERROR") && businessHeader.value().length() > 0;
    }
}

将该类打包成jar放入Connect的插件目录，然后在配置中指定自定义谓词的全类路径即可使用。

三、注意事项 #

• 确保Log4j2 KafkaAppender使用的Kafka客户端版本与Connect版本兼容，避免消息头格式不兼容问题。
• Kafka消息头的值默认是字节数组，在Connect中处理时要注意编码转换（比如用UTF-8解析）。
• 消息头总大小默认受Kafka的max.header.bytes限制（默认4KB），如果需要传递大体积元数据，建议调整该参数或改用消息体存储。

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