如何使用Kafka Streams基于消息Header过滤记录？

针对你遇到的Kafka Streams基于消息Header过滤记录的需求，以下是两种可行的实现方式：

方法一：使用Processor API（process方法） #

自定义Processor类，通过上下文转发符合条件的记录，不符合的直接丢弃。核心是利用ProcessorContext.forward()方法控制记录是否流入下游：

KStream<String, byte[]> kstream = ...; // 初始化你的输入KStream

// 绑定自定义处理器
kstream.process(() -> new CustomClassFilterProcessor());

class CustomClassFilterProcessor implements Processor<String, byte[], String, byte[]> {
    private ProcessorContext<String, byte[]> context;

    @Override
    public void init(ProcessorContext<String, byte[]> context) {
        this.context = context; // 保存上下文用于转发记录
    }

    @Override
    public void process(Record<String, byte[]> record) {
        Header classHeader = record.headers().lastHeader("class");
        // 检查Header存在且值匹配目标班级（注意用equals做字符串比较）
        if (classHeader != null && "v".equals(new String(classHeader.value()))) {
            context.forward(record); // 仅转发符合条件的记录
        }
        // 不符合条件的记录不转发，自动被过滤
    }

    @Override
    public void close() {
        // 可选：添加资源清理逻辑
    }
}

方法二：使用transformValues（DSL风格） #

如果你偏好DSL语法，可以使用transformValues（非废弃API），通过访问ProcessorContext获取Headers，返回null来过滤记录：

KStream<String, byte[]> kstream = ...;

KStream<String, byte[]> filteredStream = kstream.transformValues(() -> 
    new ValueTransformerWithKey<String, byte[], byte[]>() {
        private ProcessorContext<String, byte[]> context;

        @Override
        public void init(ProcessorContext<String, byte[]> context) {
            this.context = context;
        }

        @Override
        public byte[] transform(String key, byte[] value) {
            Headers headers = context.headers();
            Header classHeader = headers.lastHeader("class");
            // 符合条件返回原value，否则返回null触发过滤
            return (classHeader != null && "v".equals(new String(classHeader.value()))) 
                ? value 
                : null;
        }

        @Override
        public void close() {
            // 可选：添加资源清理逻辑
        }
    }
);

// 将过滤后的流写入输出Topic
filteredStream.to("output-topic");

关键注意事项 #

• 字符串比较必须用equals()而非==，避免因引用地址不同导致的判断错误
• 必须处理Header不存在的场景，防止NullPointerException
• Kafka Streams 3.0+版本中，transformValues是推荐的非废弃API，替代了旧版的transform方法

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