Kafka KStream-KTable加入竞赛条件

Question

我有以下资料：

KTable<Integer, A> tableA = builder.table("A");
KStream<Integer, B> streamB = builder.stream("B");  

streamB中的消息需要使用表A中的数据进行丰富。

示例数据：

Topic A: (1, {name=john})
Topic B: (1, {type=create,...}), (1, {type=update,...}), (1, {type=update...})

在一个完美的世界里，我想做什么

streamB.join(tableA, (b, a) -> { b.name = a.name; return b; })
       .selectKey((k,b) -> b.name)
       .to("C");

不幸的是，这对我不起作用，因为我的数据是这样的：每次将消息写入主题a时，相应的消息也会写入主题B（源是单个DB事务）。现在，在这个初始“创建”事务之后，主题B将继续接收更多消息。有时，主题B上会出现每秒数个事件，但对于给定的键，也可能出现连续事件间隔数小时的情况。

简单的解决方案不起作用的原因是原始的“创建”事务会导致竞争条件：主题A和主题B几乎同时获取它们的消息，如果主题B的消息首先到达拓扑的“连接”部分（比方说在主题A之前几毫秒消息到达那里）的tableA将不包含相应的条目。此时事件丢失。我可以在主题C上看到这种情况：一些事件显示出来，一些没有（如果我使用leftJoin，所有事件都显示出来，但有些事件有空键，这相当于丢失）。这只是初始‘创造’交易的问题。之后，每当一个事件到达主题B时，相应的条目就存在于表A中。

所以我的问题是：你如何解决这个问题？

我目前的解决方案很难看。我所做的是创建一个“B集合”，并使用

B.groupByKey()
 .aggregate(() -> new CollectionOfB(), (id, b, agg) -> agg.add(b));
 .join(tableA, ...);

现在我们有一个KTable-KTable连接，它不容易受到这种竞争条件的影响。我认为这是“丑陋的”的原因是因为在每次连接后，我必须向主题B发送一条特殊的消息，基本上说“从集合中删除我刚刚处理的事件”。如果此特殊消息没有发送到主题B，则集合将继续增长，并且集合中的每个事件将在每个连接上报告。

目前我正在研究窗口联接是否有效（将a和B读入KStreams并使用窗口联接）。我也不确定这是否有效，因为窗口大小没有上限。我想说，“窗口在‘之前’1秒开始，在‘之后’无限秒结束”。即使我能以某种方式做到这一点，我还是有点担心拥有一个无界窗口的空间需求。

如有任何建议，将不胜感激。

Answer 1

不确定您使用的是哪个版本，但最新的Kafka 2.1改进了流表联接。即使在第2.1节之前，以下内容仍然适用：

• 流表连接基于事件时间
• Kafka Streams基于事件时间处理消息，但是，以偏移顺序（对于两个输入流，首先处理具有较小记录时间戳的流）
• 如果要确保表首先更新，表更新记录的时间戳应该小于流记录

自2.1起：

• 为了允许一些延迟，您可以配置max.task。闲置的ms用于在只有一个输入主题具有输入数据的情况下延迟处理的配置

事件时处理顺序在2.0和更早版本中实现为最佳努力，这可能导致您描述的竞争条件。在2.1中，处理顺序是有保证的，并且只有当max.task.idle.ms命中时才可能被违反。

有关详细信息，请参阅https://cwiki.apache.org/confluence/display/KAFKA/KIP-353：改进Kafka Streams时间戳同步