Kafka Streams Window Store Retain Duplicates

Question

The docs are unclear. When would I want to set retain duplicates to false/true. What is this used for? Is it for something particular in RocksDB?

https://kafka.apache.org/21/javadoc/org/apache/kafka/streams/state/Stores.html#persistentWindowStore-java.lang.String-java.time.Duration-java.time.Duration-boolean-

Digging through streams internal code seems to being used to set some sequence number?

RocksDBWindowStore.java

private void maybeUpdateSeqnumForDups() {
    if (this.retainDuplicates) {
      this.seqnum = this.seqnum + 1 & 2147483647;
    }

Yes. If retainDuplicates is set, then the key is replaced by the pair (key, seqnum) before data is stored in the state store. See here: github.com/apache/kafka/blob/…. It might be used to keep all updates for an entity in the store. — user152468

Matthias J. Sax Matthias J. Sax · Accepted Answer · 2019-04-15T04:07:59

Well, as the name indicates, you can enable storing duplicates if you want to store multiple rows, with the same key. For window stores, the key is comprise of record key and window start timestamp.

Kafka Streams uses this feature for KStream-KStream joins. For this case, each input record is stored in its own window in the store (using the record timestamp as window start timestamp). Because there might be multiple records with the same key and same timestamp, it's required to enable this flag to compute the correct join. Otherwise, the join result might be incomplete.

Kafka Streams Window Store Retain Duplicates

1 Answers