Kafka在生产者上流自定义标题

Question

我有一个主题中的多个事件，我试图在这些步骤中处理：

1. 根据标题值过滤事件
2. 应用反序列化程序
3. 按键分组
4. 聚合以生成新的KTable
5. 新KTable将以流式传输方式传输到与具有新标题的新事件相同的主题

我可以使用transformValues访问标题，但不确定在执行toStream时如何注入新的标题值。

streamsBuilder.stream("my-topic")
.transformValues(new Transformer())//access headers here n filter few events
.groupByKey(Serialized.with(Serdes.String(),null)
.aggregate(()->my avro object initialization,(key,value,aggregate)->newValue(Value,aggregate),Materialized.as("my-store").withKeySerde(Serdes.String()).withValueSerde(Serdes.ByteArray())
.mapValues((key,value)->convert to bytes).toStream()

注意：我是KStream的新手。

Answer 1

可以使用处理器API添加自定义头。以与访问头相同的方式实现process方法。

new Processor() { 
    ......
   @override
   public void process(String key, String value) {
       // add a header to the elements
       context().headers().add.("key", "key");
   }
   ...
}