kafka streams会话窗口保留时间

Question

我们正在使用Kafka流的会话窗口来聚合相关事件的到达。除了聚合之外，我们还使用till（）API指定窗口的保留时间。流信息：
会话窗口（非活动时间）为1分钟，传递到until（）的保留时间为2分钟。我们使用定制的TimestampExtractor来映射事件的时间。

示例：
事件：e1；事件时间：上午10:00:00；到达时间：下午2点（同一天）
事件：e2；事件时间：上午10:00:30；到达时间：下午2:10（同一天）
第二个事件的到达时间是e1到达后10分钟，超过了保留时间非活动时间。但是旧事件e1仍然是聚合的一部分，尽管保留时间是2分钟。

问题：
1）kafka streams如何使用until（）API清理状态存储？因为指定为参数的保留值是“窗口将保持多长时间的下限”什么时候清除窗口？

2）是否有一个后台线程定期清理状态存储？如果是，那么有没有一种方法来识别清除窗口的实际时间。

3） 在保留时间后清除窗口数据的任何流配置。

Answer 1

在我回答你的具体问题之前：请注意，保留时间不是基于系统时间，而是基于“流时间”。“流时间”是一个内部跟踪的时间进度，基于时间戳提取器返回的任何内容。没有太多细节：对于您有2条记录的示例，当第二条记录到达时，“流时间”将提前30秒，因此保留时间还没有过去。

还要注意，如果没有新数据到达（至少对于一个分区），流时间不会提前。这适用于Kafka 0.11.0及更旧版本，但在未来的版本中可能会更改。

更新：Kafka 2.1更改了流时间的计算，即使一个分区不提供数据，流时间也可能提前。有关详细信息，请参阅KIP-353：改进Kafka流时间戳同步

对于您的问题：

（1） Kafka Streams将所有存储更新写入changelog主题和本地RocksDB存储。两者都被划分为具有一定大小的所谓部分。如果新数据到达（即“流时间”进展），将创建新的段。如果发生这种情况，如果旧段中的所有记录都早于保留时间（即，记录时间戳小于“流时间”减去保留时间），则删除旧段。

（2） 因此，没有后台线程，但清理是常规处理的一部分，

（3）没有强制清除旧记录/窗口的配置。

如果所有记录都已过期，则整个段将被删除，因此段内较旧的记录（最有可能具有较小/较旧的时间戳）的保留时间将长于保留时间。这种设计背后的动机是性能：按记录过期成本太高。