Kafka Streams持久存储清理

Question

是否需要一些明确的清理来防止每个持久存储的大小过大？我目前正在使用它来计算DSL API中的聚合。

Answer 1

我们遇到了类似的问题，我们只是安排了一项清洁商店的工作在我们的处理器/转换器中。只要实现你的isDataOld(nextValue(就可以了。

@Override
public void init(ProcessorContext context) {
this.kvStore = (KeyValueStore<Key, Value>) this.context.getStateStore("KV_STORE_NAME");
this.context.schedule(60000, PunctuationType.STREAM_TIME, (timestamp) -> {
KeyValueIterator<Key, Value> iterator = kvStore.all();
while (iterator.hasNext()){
KeyValue<Key,Value> nextValue = iterator.next();
if isDataOld(nextValue)
kvStore.delete(nextValue.key);
}
});
}

Answer 2

关于：

final var store =  Stores.persistentSessionStore(MATERIAL_STORE_NAME, Duration.ofDays(10));
return Stores.sessionStoreBuilder(store,
Serdes.Long(),
keySpecificAvroSerde
);