Spring和Kafka：加入3个Kafka主题生成输出Kafka流

Question

我有一个要求加入3个Kafka主题。前两个主题A和B将使用inner join添加，因为消息键相同，并且生成一个POJO与B相同的新Kafka流。现在，使用这个累积的流，我需要加入另一个主题C，并且我需要根据C中存在的字段对输出进行分组。

到目前为止，我有以下方法：

前两个主题（A和B）的KStream-KStream inner join是否可以不发布任何主题的累积输出，并且仍然可以在下面使用它

KStream-KStream（在累计流和主题C之上）

你能建议一个更好的方法或任何例子，我可以看看类似的实现在Java。

Answer 1

可以使用两个连续的联接：

KStream streamAB = streamA.join(streamB, ...);
// either:
KStream streamABC = streamA.selectKey(...) // set to the key as in streamC
                           .join(streamC, ...);
// or:
KStream streamCNew = streamC.selectKey(...); // set to the key as in streamAB
KStream streamABC = streamA.join(streamCnew, ...);
// or:
KStream streamCNew = streamC.selectKey(...); // set to a new join key
KStream streamABC = streamA.selectKey(...) // set to a new join key
                           .join(streamC, ...);

streamABC.selectKey(/* extract grouping field and set as key */).to("outputTopic");