如何在Kafka Streams响应记录中设置Correlation ID Header

要配合Spring-Kafka的ReplyingKafkaTemplate完成请求-响应的关联，核心是从请求记录中提取原Correlation ID，再将其设置到响应记录的Header中，具体步骤如下：

1. 从请求记录中提取Correlation ID #

ReplyingKafkaTemplate会自动给发送的请求记录添加KafkaHeaders.CORRELATION_ID（键值为correlationId）的Header，所以在Kafka Streams处理请求时，需要从消费到的ConsumerRecord里提取这个值。

可以通过transformValues操作获取ProcessorContext，进而访问记录的Headers：

KStream<String, RequestData> requestStream = builder.stream("request-topic");

// 转换流，携带Correlation ID和处理后的响应数据
KStream<String, Pair<String, ResponseData>> streamWithCorrelation = requestStream.transformValues(() ->
    new ValueTransformerWithKey<String, RequestData, Pair<String, ResponseData>>() {
        private ProcessorContext context;

        @Override
        public void init(ProcessorContext context) {
            this.context = context;
        }

        @Override
        public Pair<String, ResponseData> transform(String key, RequestData request) {
            // 提取Correlation ID
            Header corrHeader = context.headers().lastHeader(KafkaHeaders.CORRELATION_ID);
            String correlationId = corrHeader != null 
                ? new String(corrHeader.value(), StandardCharsets.UTF_8) 
                : null;

            // 处理请求生成响应
            ResponseData response = processRequest(request);

            // 返回Correlation ID和响应的配对
            return new Pair<>(correlationId, response);
        }

        @Override
        public void close() {}
    }
);

注：如果Correlation ID是二进制格式（比如UUID字节数组），直接保留字节数组即可，无需转成String，避免编码损耗。

2. 将Correlation ID设置到响应记录的Header中 #

处理完请求后，需要把提取到的Correlation ID添加到响应记录的Header里，再发送到响应主题。这里有两种常用方式：

方式一：手动构建ProducerRecord发送 #

用foreach操作手动创建ProducerRecord并设置Header，适合需要自定义发送逻辑的场景：

streamWithCorrelation.foreach((key, pair) -> {
    ProducerRecord<String, ResponseData> responseRecord = new ProducerRecord<>(
        "response-topic",
        key,
        pair.getValue()
    );

    // 把Correlation ID写入Header
    if (pair.getKey() != null) {
        responseRecord.headers()
            .add(KafkaHeaders.CORRELATION_ID, pair.getKey().getBytes(StandardCharsets.UTF_8));
    }

    // 发送记录到下游
    context.forward(responseRecord);
});

方式二：用RecordHeaderProvider自动添加Header #

如果不需要复杂的发送逻辑，推荐用Kafka Streams提供的RecordHeaderProvider，配合to操作自动给输出记录加Header：

streamWithCorrelation.mapValues(Pair::getValue)
    .to("response-topic", Produced.with(
        Serdes.String(),
        responseDataSerde,
        (key, originalValue) -> {
            // originalValue是之前的Pair<String, ResponseData>
            List<Header> headers = new ArrayList<>();
            if (originalValue.getKey() != null) {
                headers.add(new RecordHeader(
                    KafkaHeaders.CORRELATION_ID,
                    originalValue.getKey().getBytes(StandardCharsets.UTF_8)
                ));
            }
            return headers;
        }
    ));

关键注意事项 #

• 必须使用KafkaHeaders.CORRELATION_ID作为Header键，除非你在ReplyingKafkaTemplate的配置中自定义了correlation id的Header名称。
• 处理Correlation ID为null的情况（虽然ReplyingKafkaTemplate会自动生成，但要避免空指针异常）。
• 确保Correlation ID的字节数据和原请求完全一致，否则ReplyingKafkaTemplate无法匹配请求与响应。

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