Kafka Streams: RocksDB TTL

Question

我知道默认的TTL设置为无限（非正数）。但是，如果我们需要在存储中保留数据最长2天，我们可以使用RocksDBConfigSetter接口实现覆盖吗？也就是说，options.setWalTtlSeconds(172800)？还是会与Kafka流内部发生冲突？

Ref: https://docs.confluent.io/current/streams/developer-guide/config-streams.html#streams-developer-guide-rocksdb-config

- Raman

---

可能是Kafka Streams - Low-Level Processor API - RocksDB TimeToLive(TTL)的重复问题。 - Jacek Laskowski

Answer 1

目前来说，这是不可能做到的。出于各种技术原因，Kafka Streams以硬编码的方式禁用了RocksDB的TTL功能。关于此事也有一张工单：https://issues.apache.org/jira/browse/KAFKA-4212 暂时而言，您可以使用窗口存储来在两天后过期旧记录。例如，您可以使用stream.groupByKey().windowedBy(...).reduce(...)，其中TimeWindow为1毫秒，"dummy" reduce只返回键的最新值。

- Matthias J. Sax

---

嗨，马蒂亚斯，谢谢。由于存储是通过低级处理器API进行增强的，请建议是否有类似于上述高级DSL的解决方法。如果没有，我知道有一个开放的工单，我计划以特定频率迭代遍历键值存储并删除它们。 - Raman

您还可以通过处理器 API 创建窗口存储而不是普通的键值存储。 - Matthias J. Sax

嗨，马蒂亚斯，我遇到了类似的问题。我在拓扑中使用了windowedBy()函数，设置了1小时的时间窗口。但是当我检查磁盘使用情况时，/tmp/kafka-streams文件夹不断增加。从5G到24小时内的20G。Rocksdb会多久删除旧数据？谢谢！ 我的代码像这样.windowedBy(TimeWindows.of(Duration.ofMinutes(60)).grace(Duration.ZERO)) .reduce((event1, event2) -> event2) - thinktwice

默认情况下，保留时间为24小时。您可以通过 reduce(..., Materialized.as(null).withRetention(...)) 进行配置。 - Matthias J. Sax