Kafka Streams state stores are "compact" by default. Is it possible to set "compact,delete" with a retention policy in a state store?
2 Answers
Yes, it is possible to configure topics with retention and compaction and Kafka Streams uses this setting for windowed KTable
s.
If you really want to set this, you can update the corresponding changelog topic config manually after it is created.
However, setting topic retention time for changelog topics deletes the data only from the topic. Data is not deleted in the local state store. State stores don't offer TTL and RocksDBs TTL setting cannot be enabled (for technical reasons that we hope to resolve eventually).
If you want to delete data cleanly, you should use tombstone messages that will delete the data from the store as well as the changelog topic (instead of using retention time).
-
So, you can just send tombstone messages to the statestore changelog topic, and then the "live" statestore will learn about them and apply them to the in-mem & RocksDB representation?– xmarJun 1, 2018 at 9:39
-
1No. You should never write to the changelog topic! This would mess up your state. If you read a topic directly as a table (
builder.table(...)
) you write the tombstone into the input topic. If you table is a result of an aggregation, your aggregator needs to returnnull
as aggregation result; this, will delete the key from the store and write a tombstone to the changelog. Jun 1, 2018 at 15:11 -
1Thinking in PAPI, store.delete(Key) for KVStore and store.remove(WindowedKey) will handle all of it, then?– xmarJun 1, 2018 at 16:26
-
In the case of a window store, I guess to delete all entries you would need to fetch(key) and loop over the iterator removing each WindowedKey entry individually, right?– xmarJun 2, 2018 at 18:01
-
1It seems we are using the term "corrupted" differently... To me "corrupted" means that you lose data -- of course, if you want to clean all your state, you can delete the local state directory and clear the changelog and restart the app -- it will start up with empty state for this case (ie, "corrupted state" in the way I use the term). Jan 18 at 23:49
If you are using the default RocksDBStore there is an option to set up CompactionStyle to FIFO:
FIFO compaction style is the simplest compaction strategy. It is suited for keeping event log data with very low overhead (query log for example). It periodically deletes the old data, so it's basically a TTL compaction style.
and then use the TTL:
A new option, compaction_options_fifo.ttl, has been introduced for this to delete SST files for which the TTL has expired. This feature enables users to drop files based on time rather than always based on size, say, drop all SST files older than a week or a month.
To actually set the FIFO you have to implement RocksDBConfigSetter and set it as configuration property: rocksdb.config.setter