如何将Kafka Record Headers转发至所有派生内部主题？

Kafka Streams默认不会把记录的Headers同步到状态存储对应的changelog Topic中，要实现这一需求，可根据你的Kafka Streams版本选择以下方案：

方案一：直接配置开启Headers转发（Kafka Streams 2.4+） #

从2.4版本开始，官方提供了原生配置支持，无需额外编码：

• 针对单个状态存储：创建存储时添加INCLUDE_HEADERS_CONFIG配置

  Stores.keyValueStoreBuilder(
      Stores.persistentKeyValueStore("my-state-store"),
      Serdes.String(),
      Serdes.String()
  ).withConfig(Map.of(StoreConfig.INCLUDE_HEADERS_CONFIG, "true"));
  

• 全局生效：在StreamsConfig中设置默认状态存储配置

  Properties props = new Properties();
  props.put(StreamsConfig.DEFAULT_STATE_STORE_CONFIG, 
            Map.of(StoreConfig.INCLUDE_HEADERS_CONFIG, "true"));
  

这是最简单高效的方式，推荐优先使用。

方案二：自定义处理器拦截状态变更（适用于低版本） #

如果你的Kafka Streams版本低于2.4，或者需要对Headers做定制化处理，可以通过包装处理器上下文来实现：

1. 实现ProcessorContextWrapper，拦截状态存储的写入操作，将当前记录的Headers注入到changelog记录中：

   public class HeaderForwardingContextWrapper implements ProcessorContextWrapper {
       private final ProcessorContext originalContext;
       private Headers currentHeaders;
   
       public HeaderForwardingContextWrapper(ProcessorContext originalContext) {
           this.originalContext = originalContext;
       }
   
       public void setCurrentHeaders(Headers headers) {
           this.currentHeaders = headers;
       }
   
       @Override
       public ProcessorContext unwrap() {
           return originalContext;
       }
   
       @Override
       public <S extends StateStore> S getStateStore(String name) {
           S store = originalContext.getStateStore(name);
           // 包装状态存储，写入时附加Headers
           if (store instanceof KeyValueStore) {
               return (S) new HeaderForwardingKeyValueStore<>((KeyValueStore) store, this::getCurrentHeaders);
           }
           return store;
       }
   
       private Headers getCurrentHeaders() {
           return currentHeaders;
       }
   }
   

2. 在自定义Processor中使用这个包装类，处理记录时同步当前Headers：

   public class MyProcessor<K, V> extends AbstractProcessor<K, V> {
       private HeaderForwardingContextWrapper wrappedContext;
       private KeyValueStore<K, V> store;
   
       @Override
       public void init(ProcessorContext context) {
           wrappedContext = new HeaderForwardingContextWrapper(context);
           super.init(wrappedContext);
           store = wrappedContext.getStateStore("my-state-store");
       }
   
       @Override
       public void process(K key, V value) {
           // 记录当前记录的Headers
           wrappedContext.setCurrentHeaders(context.headers());
           // 执行状态更新，此时changelog会带上Headers
           store.put(key, value);
       }
   }
   

这种方式需要更多编码，但能实现更灵活的Headers控制逻辑。

注意事项 #

• 开启Headers转发后，changelog Topic的数据量会有所增加，需要评估存储和性能影响。
• 确保使用的序列化器兼容Headers处理，默认序列化器无需额外调整，自定义序列化器需注意处理Headers字段。

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