如何用Kafka Stream向同一主题发送带不同Headers的多条消息

要实现拆分后每条消息携带独立Kafka Headers，核心是要在拆分消息的同时为每个片段生成自定义Headers，替代原代码中仅处理Value的逻辑。以下提供两种可行方案：

方案1：使用flatMap直接生成带Headers的记录 #

适合Kafka Streams 2.5+版本（支持带Headers的KeyValue构造），代码更简洁：

import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.streams.kstream.KeyValue;

// ...

inputStream
  .transformValues(() -> new Transformation())
  // 替换flatMapValues为flatMap，获取完整记录上下文并生成带独立Headers的消息
  .flatMap((key, value, recordContext) -> {
      List<KeyValue<String, String>> splitRecords = new ArrayList<>();
      String[] messageParts = value.split("~");
      
      for (int index = 0; index < messageParts.length; index++) {
          String part = messageParts[index];
          // 为每个拆分片段创建独立Headers实例
          Headers headers = new RecordHeaders();
          
          // 示例：按索引设置header
          headers.add("message-sequence", String.valueOf(index).getBytes());
          // 示例：根据消息内容动态设置header
          if (part.startsWith("ERR")) {
              headers.add("message-level", "error".getBytes());
          } else if (part.startsWith("WARN")) {
              headers.add("message-level", "warning".getBytes());
          } else {
              headers.add("message-level", "info".getBytes());
          }
          // 可选：继承原始记录的Headers
          // headers.addAll(recordContext.headers());
          
          splitRecords.add(KeyValue.pair(key, part, headers));
      }
      return splitRecords;
  })
  .split()
    .branch(
      (key, value) -> key.startsWith("ERR"),
      Branched.withConsumer(ks -> ks.to(errorTopic)))
    .defaultBranch(Branched.withConsumer(ks -> ks.to(outboundTopic)));

关键说明 #

• 用flatMap替代flatMapValues：可以访问完整的记录上下文，并且支持返回携带自定义Headers的KeyValue对象；
• 独立Headers实例：每个拆分片段对应全新的RecordHeaders，避免不同消息间Header污染；
• 灵活定制：可根据消息内容、索引等业务规则动态生成Header。

方案2：使用transform处理器手动设置Headers #

适合低版本Kafka Streams（不支持带Headers的KeyValue），灵活性更高：

import org.apache.kafka.streams.kstream.Transformer;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.kstream.KeyValue;

// ...

inputStream
  .transformValues(() -> new Transformation())
  .flatMapValues(value -> Arrays.asList(value.split("~")))
  // 新增transform步骤，为每条消息设置独立Headers
  .transform(() -> new Transformer<String, String, KeyValue<String, String>>() {
      private ProcessorContext context;

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

      @Override
      public KeyValue<String, String> transform(String key, String value) {
          // 清空原有Headers（按需选择保留或覆盖）
          context.headers().clear();
          
          // 根据业务逻辑设置自定义Headers
          context.headers().add("content-length", String.valueOf(value.length()).getBytes());
          if (value.contains("ERROR")) {
              context.headers().add("status", "failed".getBytes());
          } else {
              context.headers().add("status", "success".getBytes());
          }
          
          return KeyValue.pair(key, value);
      }

      @Override
      public void close() {}
  })
  .split()
    .branch(
      (key, value) -> key.startsWith("ERR"),
      Branched.withConsumer(ks -> ks.to(errorTopic)))
    .defaultBranch(Branched.withConsumer(ks -> ks.to(outboundTopic)));

关键说明 #

• 通过ProcessorContext直接操作当前消息的Headers；
• 适合复杂Header生成逻辑（如依赖外部配置、计算哈希值等）；
• 可选择清空原有Headers或在其基础上追加新Header。

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