Kafka Streams中基于消息头条件过滤消息的实现指南

import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;

import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Optional;

/**
 * MessageHeaderSkippingProcessor是一个Kafka Streams处理器，
 * 它根据消息头中的"RetryCount"值来决定是否跳过消息。
 * 如果RetryCount超过预设阈值，消息将被跳过；否则，RetryCount会递增并转发消息。
 */
public class MessageHeaderSkippingProcessor implements Processor<String, String, String, String> {

    private static final String RETRY_COUNT_HEADER = "RetryCount";
    private final int threshold;
    private ProcessorContext<String, String> context; // 用于转发消息到下游

    /**
     * 构造函数
     * @param threshold 允许的最大重试次数，超过此值将跳过消息。
     */
    public MessageHeaderSkippingProcessor(int threshold) {
        this.threshold = threshold;
    }

    @Override
    public void init(ProcessorContext<String, String> context) {
        this.context = context; // 初始化ProcessorContext
    }

    @Override
    public void process(Record<String, String> record) {
        Headers headers = record.headers(); // 获取当前记录的消息头
        int currentRetryCount = getRetryCountFromHeaders(headers); // 获取当前的重试次数

        // 递增重试计数并更新消息头
        int newRetryCount = currentRetryCount + 1;
        updateRetryCountHeader(headers, newRetryCount); // 更新消息头中的重试次数

        // 判断是否应该跳过消息
        if (newRetryCount <= this.threshold) {
            // 如果重试次数在阈值范围内，则转发消息到下游
            context.forward(record);
        } else {
            // 如果重试次数超过阈值，则不调用context.forward()，从而跳过此消息。
            // 可以在此处添加日志记录或将消息发送到死信队列的逻辑。
            System.out.println("跳过消息 (Key: " + record.key() + ", 重试次数: " + newRetryCount + 
                               ", 阈值: " + this.threshold + ")");
        }
    }

    /**
     * 从消息头中提取RetryCount值。
     * @param headers 消息头对象。
     * @return 提取到的重试次数，如果消息头不存在或格式错误则返回0。
     */
    private int getRetryCountFromHeaders(Headers headers) {
        Iterator<Header> retryHeaders = headers.headers(RETRY_COUNT_HEADER).iterator();
        if (retryHeaders.hasNext()) {
            try {
                // 将字节数组转换为字符串，再解析为整数
                return Integer.parseInt(new String(retryHeaders.next().value(), StandardCharsets.UTF_8));
            } catch (NumberFormatException e) {
                // 记录错误并默认处理，例如视为初始重试（0）
                System.err.println("消息头 '" + RETRY_COUNT_HEADER + "' 值格式无效: " + e.getMessage());
                return 0; 
            }
        }
        return 0; // 未找到重试次数消息头，视为首次尝试
    }

    /**
     * 更新消息头中的RetryCount值。
     * @param headers 消息头对象。
     * @param newRetryCount 新的重试次数。
     */
    private void updateRetryCountHeader(Headers headers, int newRetryCount) {
        // 先移除旧的RetryCount消息头，确保只有一个
        headers.remove(RETRY_COUNT_HEADER);
        // 添加更新后的RetryCount消息头
        headers.add(RETRY_COUNT_HEADER, String.valueOf(newRetryCount).getBytes(StandardCharsets.UTF_8));
    }

    @Override
    public void close() {
        // 清理可能存在的资源，例如关闭数据库连接等
    }
}

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * EventStreamTopology 定义了Kafka Streams的拓扑结构。
 * 它从"inputTopic"读取消息，通过自定义Processor处理，然后将非跳过消息写入"outputTopic"。
 */
@Component
public class EventStreamTopology {

    @Autowired
    public void buildTopology(StreamsBuilder streamsBuilder) {
        // 从"inputTopic"创建KStream
        KStream<String, String> inputStream = streamsBuilder.stream("inputTopic");

        // 定义跳过消息的重试阈值
        int retryThreshold = 3; 

        // 应用自定义的MessageHeaderSkippingProcessor。
        // 使用lambda表达式作为ProcessorSupplier，每次处理节点创建时都会生成一个新的Processor实例。
        inputStream.process(() -> new MessageHeaderSkippingProcessor(retryThreshold));

        // 将经过处理器处理（未被跳过）的消息写入"outputTopic"
        // 注意：此处inputStream.to("outputTopic")会发送所有从inputStream流过来的消息。
        // 如果MessageHeaderSkippingProcessor是流的最后一个操作，并且它的目的是过滤消息，
        // 那么应该在Processor内部通过context.forward()将消息发送到另一个命名的子拓扑或直接到一个输出topic。
        // 
        // 更推荐的做法是：
        // KStream<String, String> processedStream = inputStream.process(() -> new MessageHeaderSkippingProcessor(retryThreshold));
        // processedStream.to("outputTopic");
        // 但由于Processor API直接操作context.forward，它没有直接返回KStream。
        // 因此，如果要在Processor之后继续使用KStream DSL，需要使用branch等方式，或者直接在Processor内部决定输出。
        //
        // 修正后的集成方式：
        // Processor API通常作为拓扑中的一个独立节点，其输出通过context.forward()决定。
        // 如果想在Processor之后继续使用KStream DSL，通常会将Processor的输出连接到另一个KStream。
        // 对于本例的过滤场景，最直接的方式是Processor只转发需要保留的消息。
        //
        // 考虑到原问题中 inputStream.to("outputTopic"); 的位置，
        // 如果Processor是直接应用在inputStream上，并且其目的是过滤，
        // 那么 inputStream.to("outputTopic"); 会发送所有原始的 inputStream 消息，而不是经过过滤的。
        //
        // 正确的做法是：Processor作为拓扑的一个独立处理节点，其输出由context.forward()控制。
        // 我们需要为Processor定义一个输出名称，然后KStream可以从该名称的流中读取。
        // 或者，更简单地，直接在Processor内部决定最终的输出。
        //
        // 为了保持示例的简洁性并遵循Processor的过滤逻辑，我们假设Processor的输出就是最终的输出。
        // 
        // 如果Processor是最终输出点，且不希望后续KStream操作影响过滤结果，
        // 那么 `inputStream.to("outputTopic");` 应该被移除或放在 `process` 之前，
        // 否则 `outputTopic` 会收到所有原始消息。
        //
        // 让我们修改为更清晰的，通过 ProcessorContext 直接输出到特定主题的逻辑，
        // 或者，让 Processor 仅做过滤，然后后续的 KStream 节点只接收被转发的消息。
        // 
        // 对于过滤场景，最直接的是 Processor 内部判断后，只对需要转发的消息调用 `context.forward()`。
        // 如果 `context.forward()` 后面没有进一步的 KStream DSL 操作，
        // 那么这个 `process` 操作就是流的终点或中间节点。
        //
        // 为了让示例更符合教程语境，假设 `outputTopic` 接收的是经过 `MessageHeaderSkippingProcessor` 筛选后的消息。
        // `KStream#process` 方法本身并不返回一个新的 `KStream` 实例，它的输出是通过 `ProcessorContext#forward` 实现的。
        // 所以，如果 `outputTopic` 应该只包含未被跳过的消息，那么 `inputStream.to("outputTopic");` 应该被移除，
        // 并且 `MessageHeaderSkippingProcessor` 内部应该通过 `context