Kafka Streams - 低级处理器 API - RocksDB TimeToLive(TTL) [英] Kafka Streams - Low-Level Processor API - RocksDB TimeToLive(TTL)

问题描述

我正在尝试使用低级处理器 API.我正在使用处理器 API 对传入记录进行数据聚合，并将聚合记录写入 RocksDB.

I'm kind of experimenting with the low level processor API. I'm doing data aggregation on incoming records using the processor API and writing the aggregated records to RocksDB.

但是，我想保留在rocksdb中添加的记录仅在24小时内处于活动状态.24 小时后应删除该记录.这可以通过更改 ttl 设置来完成.但是，没有太多文档可以帮助我解决这个问题.

However, I want to retain the records added in the rocksdb to be active only for 24hr period. After 24hr period the record should be deleted. This can be done by changing the ttl settings. However, there is not much documentation where I can get some help on this.

如何更改 ttl 值?我应该使用什么 java api 将 ttl 时间设置为 24 小时，当前默认的 ttl 设置时间是多少?

how do I change the ttl value? What java api should I use to set the ttl time to 24 hrs and whats the current default ttl settings time?

推荐答案

我相信目前还没有通过 api 或配置公开.RocksDBStore 在打开 RocksDB 时传递硬编码的 TTL:https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java#L158并且硬编码值只是 TTL_SECONDS = TTL_NOT_USED (-1)(请参阅同一文件中的第 79 行).

I believe this is not currently exposed via the api or configuration. RocksDBStore passes a hard-coded TTL when opening a RocksDB: https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java#L158 and the hardcoded value is simply TTL_SECONDS = TTL_NOT_USED (-1) (see line 79 in that same file).

目前有 2 个关于在 state store 中公开 TTL 支持的公开票:KAFKA-4212 和 KAFKA-4273:https://issues.apache.org/jira/issues/?jql=project%20%3D%20KAFKA%20AND%20text%20~%20%22rocksdb%20ttl%22我建议您对描述您的用例的其中一个发表评论，以推动他们向前发展.

There are currently 2 open ticket regarding exposing TTL support in the state stores: KAFKA-4212 and KAFKA-4273: https://issues.apache.org/jira/issues/?jql=project%20%3D%20KAFKA%20AND%20text%20~%20%22rocksdb%20ttl%22 I suggest you comment on one of them describing your use case to get them moving forward.

在此期间，如果您现在需要 TTL 功能，状态存储是可插入的，并且 RocksDBStore 源随时可用，因此您可以分叉它并设置您的 TTL 值(或者，像与 KAFKA-4273 相关的拉取请求建议，从配置中获取).

In the interim, if you need the TTL functionality right now, state stores are pluggable, and the RocksDBStore sources readily available, so you can fork it and set your TTL value (or, like the pull request associated with KAFKA-4273 proposes, source it from the configs).

我知道这并不理想，真诚地希望有人提出更令人满意的答案.

I know this is not ideal and sincerely hope someone comes up with a more satisfactory answer.

这篇关于Kafka Streams - 低级处理器 API - RocksDB TimeToLive(TTL)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！