KStream加入重试/延迟的KStream

Question

我们正在尝试实现下面描述的用例,我们有我们希望克服的实现问题,

用例,

我们试图通过匹配两个流的消息中的键（JSON）来实现两个Kafka主题之间的KStream连接。此外，我们还应该维护消息序列，因为它是从源代码到达KStream的。

场景是，如果匹配的键还没有到达任何一个流，我们应该停止或重试join，直到预期的键到达其他主题。我们想把不匹配的记录放回KStream，但在这种情况下，序列没有保证。

问题1：如何在其他主题中停止或保持join直到预期的键到达。例如，KTable有密钥100，但是KStream还没有收到密钥100，那么我们应该重试Join或者保持KStream直到密钥100到达。

在KTable和KStream之间的连接上，Java更好，就像我们做的POC一样

KTable<String, String> leftStream = builder.table("stream1");
KStream<String, String> rightStream = builder.stream("stream2");
KStream<String, String> outstream = rightStream.leftJoin(leftStream, (orig_msg, description) -> {
         String new_msg = "";
            if (description != null) {
                  new_msg = orig_msg+"-->Matched--"+description;
            }else {
                new_msg = orig_msg+"-->UnMatched<--"+description;
            }
                return new_msg;
     });

Answer 1

在示例中，您正在执行KStream-to-KTable的左联接。Kafka Streams join语义指定：(a)只有到达KStream的数据才会触发join输出，并且(b)如果当新的KStream事件到达时，KTable（join的右侧）中没有匹配的数据，那么仍然会立即产生一个join输出，但表端数据的输出为null（即，KTable端不会等待数据到达）。

问题1：如何在其他主题中停止或保持join直到预期的键到达。例如，KTable有密钥100，但是KStream还没有收到密钥100，那么我们应该重试Join或者保持KStream直到密钥100到达。

首先，不能使用内置的Kafka Streams功能停止或保留连接。

其次，您提供的特定示例在实践中不会发生，因为（参见上文）到达KTable的事件不会产生连接输出。只有当事件到达KStream时，才会(a)查找KTable和(b)生成连接输出，而不管(a)的结果如何。

但是在KStream-KTable左联接中可能发生的是相反的例子：KStream有密钥100，但是KTable尚未接收密钥100。这个怎么处理？见下文。

问题2：有没有办法在KStream（延迟的KStream）中放入延迟或间隔来接收有延迟时间或间隔的消息。

有一个示例应用程序演示了这一点，恰巧与上面的用例类似：请参阅https://github.com/confluentinc/kafka-streams-examples（直接链接到ConfluentV5.2.1/Apache Kafka 2.2的CustomStreamTableJoin示例）的CustomStreamTableJoin。