2017-05-09 107 views
1

我正在尝试低级别处理器API。我正在使用处理器API对传入记录进行数据聚合,并将聚合记录写入RocksDB。Kafka Streams - 低级别处理器API - RocksDB TimeToLive(TTL)

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

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

回答

1

我相信这目前不通过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行)。

目前有2关于暴露在国营商店TTL支持公开售票:卡夫卡4212和卡夫卡4273: https://issues.apache.org/jira/issues/?jql=project%20%3D%20KAFKA%20AND%20text%20~%20%22rocksdb%20ttl%22 我建议你在其中一人描述你的使用情况,让他们前进的评论。

在此期间,如果您现在需要TTL功能,状态存储是可插入的,并且RocksDBStore资源随时可用,因此您可以将其分配并设置TTL值(或与KAFKA- 4273建议,从配置中获取它)。

我知道这并不理想,真诚希望有人提出一个更满意的答案。