Kafka Streams KeyValueStore retention.bytes

Question

我和KeyValueStore有一个有趣的行为，我有一些假设来解释它，也许你可以告诉我是对的还是错的……

我配置了状态存储，如下所示

Map<String, String> storeConfig = new HashMap<>();
storeConfig.put(TopicConfig.RETENTION_MS_CONFIG, TimeUnit.DAYS.toMillis(30));
storeConfig.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete");

StoreBuilder store1 = Stores.keyValueStoreBuilder(
   Stores.persistentKeyValueStore("STORE1"),
   Serdes.String(),
   Serdes.String()
);

streamsBuilder.addStateStore(store1.withLoggingEnabled(storeConfig));

在这种配置下，我预计超过30天的数据集将消失，但我观察到的是完全不同的东西。

当我查看商店的rockdb目录时，它每14451个字节就会滚动一次文件，我在目录中就有这样的结构

14451  1. Oct 19:00 LOG
14181 30. Sep 15:59 LOG.old.1569854012833395
14451 30. Sep 17:40 LOG.old.1569918431235734
14451  1. Oct 11:05 LOG.old.1569949239434224

它似乎没有实现配置的超过30天的保留期，它还实现了超过文件大小。

我在互联网上发现还有参数Topic.RETENTION_BYTES_CONFIG 'retention.bytes'，我是否也需要配置这个参数，这样我的数据在保留期间是可见的，并且不会因为文件大小而被删除(我知道我的密钥有值，但发生这种现象后我无法访问它)……

答案Thx ..

Answer 1

在内部，KeyValueStores使用RocksDB，而RocksDB在内部使用所谓的LSM-树(日志结构合并树)，它创建了许多较小的段，这些段稍后会组合成较大的段。在这个“压缩”步骤之后，可以删除较小的段文件，因为数据被复制到较大的段文件中。因此，没有什么好担心的。

此外，Topic.RETENTION_MS_CONFIG是一个主题配置，与Kafka Streams应用程序的本地存储无关。此外，KeyValueStore将永远保留数据，直到通过"tombstone“消息显式删除。因此，如果为基础changelog主题设置保留时间，则可能会删除主题中的数据，但不会删除本地存储区中的数据。

如果要将保留时间应用于本地存储区，则不能使用KevValueStore，但可以使用支持保留时间的WindowedStore。