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.Optional;

/**
 * 自定义Processor，用于根据消息头中的RetryCount值实现条件跳过。
 */
public class MessageHeaderProcessor implements Processor<String, String, String, String> {

    public static final String RETRY_COUNT_HEADER = "RetryCount";
    private final Integer threshold;
    private ProcessorContext<String, String> context; // 保存ProcessorContext实例

    /**
     * 构造函数，传入重试次数阈值。
     * @param threshold 允许的最大重试次数。
     */
    public MessageHeaderProcessor(Integer threshold) {
        this.threshold = threshold;
    }

    /**
     * 初始化Processor，获取并保存ProcessorContext。
     * @param context Processor上下文。
     */
    @Override
    public void init(ProcessorContext<String, String> context) {
        this.context = context;
    }

    /**
     * 处理每个传入的记录。
     * 根据消息头中的RetryCount值，决定是否将消息转发到下游。
     * @param record 待处理的Kafka记录。
     */
    @Override
    public void process(Record<String, String> record) {
        Headers headers = record.headers();
        int currentRetryCount = 0;

        // 尝试获取并解析RetryCount头
        Optional<Header> retryCountHeaderOpt = Optional.ofNullable(headers.lastHeader(RETRY_COUNT_HEADER));
        if (retryCountHeaderOpt.isPresent()) {
            try {
                currentRetryCount = extractRetryCount(retryCountHeaderOpt.get().value());
            } catch (NumberFormatException e) {
                // 处理解析错误，例如记录日志，并将其视为0或默认值
                System.err.println("Error parsing RetryCount header for key: " + record.key() + ". Error: " + e.getMessage());
            }
        }

        // 更新或添加RetryCount头
        headers.remove(RETRY_COUNT_HEADER); // 移除旧的，准备添加新的
        int newRetryCount = currentRetryCount + 1;
        headers.add(RETRY_COUNT_HEADER, String.valueOf(newRetryCount).getBytes(StandardCharsets.UTF_8));

        // 判断是否超过阈值，决定是否转发消息
        if (newRetryCount <= this.threshold) {
            // 如果未超过阈值，则将消息转发到下游
            context.forward(record);