我有一个与StateStore交互的处理器,用于过滤并对消息执行复杂的逻辑.在process(key,value)我context.forward(key,value)用来发送我需要的键和值的方法中.出于调试目的,我也打印出来.
我有一个KStream mergedStream,它来自另外两个流的连接.我想将处理器应用于该流的记录.我实现了这个:mergedStream.process(myprocessor,"stateStoreName")
当我启动这个程序时,我可以看到要打印到我的控制台的正确值.但是,如果我使用主题上mergedStream.to("topic")的值将mergedStream发送到主题,则不是我在处理器中转发的那个,而是原始的.
我使用kafka-streams 0.10.1.0.
将我在处理器中转发的值转换为另一个流的最佳方法是什么？
是否可以将Processor API与KStream DSL创建的流混合使用？