Kafka Streams DSL：如何关闭由流分支/分组创建的子流？

Question

我正在用Scala编写一个Kafka Streams应用程序，我担心潜在的内存泄漏/总的资源使用。

是否有一种方法向Kafka发出信号，让它“关闭”分组/分支操作创建的特定子流，并释放相关资源？

为了演示潜在的问题，让我们考虑一个电子商务应用程序，它将订单状态更改事件推送到一个名为“my-super-input-topic”的Kafka主题。每个订单都由OrderId唯一标识，OrderId用作Kafka消息键。

假设我们需要计算每个订单的状态更新计数，并将结果推送到“my-super-output-topic”主题。下面的代码段演示了如何在Scala中执行此操作：

// ...
val builder = new StreamsBuilder
val ktable = builder.stream("my-super-input-topic")
    .groupByKey
    .count

ktable.toStream.to("my-super-output-topic")
// ...

据我所知，.groupby/.groupbykey将源流划分为N个子流（在本例中，每个订单一个子流）。上面的代码没有指定任何保留窗口，因此，即使给定的订单（子流）在几小时不活动后接收到事件--它仍将被正确处理，并且更新将被推送到包含正确聚合计数的接收器主题。

因此，我得出结论，Kafka将每个子流的信息保存在某种内部存储中。

然而，订单有有限的生存期，并且在某个时间订单完成之后，这意味着与该订单相关的子流将永远不会接收到更多的事件。但Kafka仍然把它当作一个有效的，等待进一步的消息，越来越多的“死”子流会随着越来越多的订单被完成而累积。如果Kafka至少专用一些资源来跟踪每个子流，那么“死的”子流就会造成广泛的内存使用，即使这是完全不必要的。

因此，一旦系统了解到相关订单已完成，处理/关闭特定子流将是合理的。

注意：这是一个虚构的用例来演示具体问题，而不是一个真实的任务。请不要建议在没有Kafka流的情况下实施。

Answer 1

您的聚合将永远为每个键保留一个计数，这是正确的。然而，“子流”是基于每个分区的，因此，每个子流应该总是包含一些数据。

无法关闭拓扑的部分。

如果您担心KTable存储区的无限增长，您可以考虑（1）使用窗口存储区，该存储区最终将驱逐旧数据；（2）使用Aggregate()而不是count：默认情况下Aggregate()将仅计数，但如果订单已完成，UDF可以返回NULL--这将从存储区中删除订单的键值对。（3）或者您考虑不使用DSL，而是使用处理器API，该API提供更多的控制/灵活性来主状态存储（您也可以考虑使用“标点符号”）。

您可能还对以下内容感兴趣：https://issues.apache.org/jira/browse/kafka-4212