Status

Current state: Accepted

Discussion thread: here

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Kafka-streams applications usually consume from topics which are written with at-least-once semantics. In companies where Kafka is the backbone platform, an application maintained by team A may read topics that are written by team B. While team A may expect that the topic is written with exactly-once semantics, team B may not always fulfill that requirement. Thus, team A is forced to add a custom processor where they deduplicate data. This can result in having to write a custom deduplication processor for every external topic, and for every Kafka-streams application.

To save applications from having to write a deduplication processor each time, we introduce a new deduplication api that does the job for them.

Public Interfaces

public interface KStream<K, V> {  

    /**
     * Filter out duplicates from this stream based on record's key, within the provided time interval.
     * After receiving a non-duplicate record, any record with the same key that is received within the provided deduplicationInterval
     * (interval bounds are inclusive) will be discarded. This applies to both in-order and out-of-order duplicates (see example below).
     * After deduplicationInterval has elapsed since a non-duplicate record is received, a new record having
     * the same key is considered a non-duplicate, and is forwarded to the resulting {@link KStream}.
     * <p>
     * A late record that is late by strictly more than deduplicationInterval from the current stream time
     * is systematically forwarded, unless it had a forwarded duplicate (within its deduplicationInterval)
     * whose timestamp is in the window [currentStreamTime-deduplicationInterval, currentStreamTime] (see example below).
     * Records with a {@code null} key are always forwarded to the resulting {@link KStream}, no deduplication is performed on them.
     * <p>
     * In the following example, events r1 to r6 have the same key.
     * <pre>{@code
     * r1 at t -> forwarded
     * r2 at t+deduplicationInterval -> discarded
     * r3 at t-deduplicationInterval -> discarded
     * r4 at t+deduplicationInterval+1 -> forwarded
     * r5 at t -> forwarded (late record)
     * r6 at t -> forwarded (late record)
     * }</pre>
     * In the following example, we consider events having different keys k1 and k2.
     * <pre>{@code
     * k1 at t -> forwarded
     * k2 at t+deduplicationInterval -> forwarded (currentStreamTime = t+deduplicationInterval)
     * k1 at t-deduplicationInterval -> discarded (although this is a late event, it has a duplicate which is in the window [currentStreamTime-deduplicationInterval, currentStreamTime]
     * }</pre>
     * <p>
     * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)},
     * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)} or
     * {@link #process(ProcessorSupplier, String...)}) an internal repartitioning topic will be created in Kafka.
     * This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
     * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
     * <name> is an internally generated name, and "-repartition" is a fixed suffix.
     *
     * @param deduplicationInterval             the duration within which subsequent duplicates of a record will be discarded
     * @return                                  a KStream that contains the same records of this KStream without duplicates
     */
    KStream<K, V> deduplicateByKey(final Duration deduplicationInterval);


    KStream<K, V> deduplicateByKey(final Duration deduplicationInterval,
                                   final Deduplicated<K, V> deduplicated);


    /**
     * Filter out duplicates from this stream based on record's key and the provided {@link KeyValueMapper}, within the provided time interval.
     * The provided {@link KeyValueMapper} maps a record to an id.
     * For two records to be duplicate, they must have the same key and the same id. 
     * <p>
     * Example of usage.
     * <pre>{@code
     * KStream<String, Object> inputStream = builder.stream("topic");
     *
     * KStream<String, Object> outputStream = inputStream.deduplicateByKeyValue(new KeyValueMapper<String, Object, String>() {
     *     String apply(String key, Object value) {
     *         return value.id;
     *     }
     * }, Duration.ofSeconds(60));
     * }</pre>
     * </p>
     * After receiving a non-duplicate record, any duplicates received within the provided deduplicationInterval
     * (interval bounds are inclusive) will be discarded. This applies to both in-order and out-of-order duplicates (see examples below).
     * After deduplicationInterval has elapsed since a non-duplicate record is received, a new record having
     * the same (key, id) is considered a non-duplicate, and is forwarded to the resulting {@link KStream}.
     * <p>
     * A late record that is late by strictly more than deduplicationInterval from the current stream time
     * is systematically forwarded, unless it had a forwarded duplicate (within its deduplicationInterval)
     * whose timestamp is in the window [currentStreamTime-deduplicationInterval, currentStreamTime] (see example below).
     * Records with a {@code null} key OR a {@code null} id are always forwarded to the resulting {@link KStream}, no deduplication is performed on them.
     * <p>
     * In the following example, events r1 to r6 have the same key and id.
     * <pre>{@code
     * r1 at t -> forwarded
     * r2 at t+deduplicationInterval -> discarded
     * r3 at t-deduplicationInterval -> discarded
     * r4 at t+deduplicationInterval+1 -> forwarded
     * r5 at t -> forwarded (late record)
     * r6 at t -> forwarded (late record)
     * }</pre>
     * In the following example, we consider events having different id k1 and k2.
     * <pre>{@code
     * k1 at t -> forwarded
     * k2 at t+deduplicationInterval -> forwarded (and currentStreamTime = t+deduplicationInterval)
     * k1 at t-deduplicationInterval -> discarded (although this is a late event, it has a duplicate which is in the window [currentStreamTime-deduplicationInterval, currentStreamTime]
     * }</pre> 
     * <p>
     * If a key changing operator was used before this operation (e.g., {@link #selectKey(KeyValueMapper)},
     * {@link #map(KeyValueMapper)}, {@link #flatMap(KeyValueMapper)} or
     * {@link #process(ProcessorSupplier, String...)}) an internal repartitioning topic will be created in Kafka.
     * This topic will be named "${applicationId}-<name>-repartition", where "applicationId" is user-specified in
     * {@link StreamsConfig} via parameter {@link StreamsConfig#APPLICATION_ID_CONFIG APPLICATION_ID_CONFIG},
     * <name> is an internally generated name, and "-repartition" is a fixed suffix.
     *
     * @param idSelector                        a {@link KeyValueMapper} that returns the unique id of the record
     * @param deduplicationInterval             the duration within which subsequent duplicates of a record will be discarded
     * @param <KR>                              the type of the deduplication id
     * @return                                  a KStream that contains the same records of this KStream without duplicates
     */
    <KR> KStream<K, V> deduplicateByKeyValue(final KeyValueMapper<? super K, ? super V, ? extends KR> idSelector,
                                             final Duration deduplicationInterval);


    <KR> KStream<K, V> deduplicateByKeyValue(final KeyValueMapper<? super K, ? super V, ? extends KR> idSelector,
                                             final Duration deduplicationInterval,
                                             final Deduplicated<KR, V> deduplicated);

}

...

• control the time interval within which duplicates of a record are discarded
• limit storage by preventing the state store from growing indefinitely

Proposed changes

As mentioned in the java docs, we propose both deduplicating based on record key or based on a (computed) id.

...

Finally, given two duplicate records, the first record received (i.e. the one with the least offset) is the one forwarded.

Semantics

Null deduplication key

For `deduplicateByKey()`, records whose key is null are always forwarded to the output stream.

...

• if the id is null: we don't want to consider only the key for deduplication (because other records with same key were not considered duplicate). The user can still map the null id to a constant value if he wants this behavior.
• if the key is null: we don't want a non-deterministic behavior. Two records with a null key may or may not be part of the same partition. In the first case they would be deduplicated, in the second case they would not be. The user can map the key to a constant value beforehand if he wants to deduplicate such records.

Duplicate sequence

Example: deduplicationInterval = 10s. Following events a1, a2 & a3 are duplicates.

...

• event a1 at t → Forwarded and saved in the store
• event a2 at t-8s → Not forwarded, we don't update the store
• event a3 at t-11s → Forwarded (see section Late events)

Deduplication interval boundaries

Deduplication interval ends are inclusive.

...

• event a1 @t=5s → Forwarded
• event a2 @t=5s → Dropped
• event a3 @t=6s → Forwarded

Late events by more than deduplicationInterval

Late events which are more than deduplicationInterval in the past (with respect to stream time) cannot be saved in the store (they will be constantly purged), hence their duplicates received afterwards won't be detected as duplicates. We have two choices:

...

• event k1 @t=10s → Forwarded    
• event k2 @t=21s → Forwarded       → event k1 @t=10s is now purged from the store
• event k1 @t=9s → Forwarded      (late event with no duplicate in the store)

Compatibility, Deprecation, and Migration Plan

The proposed change is backwards compatible, no deprecation or migration needed.

Rejected Alternatives

Nature of the underlying store

• We use a window store to save the ids "seen" by the processor. Setting the retention of this store to deduplicationInterval allows to automatically purge old records.

Discarding late events

• Out-of-order events that are within deduplicationInterval from max observed streamTime are handled by the processor's logic above.
• Image RemovedImage AddedHowever, the above processor's logic doesn't handle events that are more than deduplicationInterval in the past (i.e. with respect to maxObservedStreamTime).Image Removed Image Added In fact, after receiving such a record, it won't be saved in the window store because it is considered expired. Any duplicate to it that arrives afterwards would therefore not be detected as duplicate.

...

          Of course, the user can set deduplicationInterval to whatever value he judges to be sufficiently large.

Processor logic

• check if the record is more than deduplicationInterval late
  • If so, ignore it.
• evaluate the record's deduplication id using the supplied KeyValueMapper
• fetch entries in the store having this key within the time interval(record.timestamp-deduplicationInterval, record.timestamp+deduplicationInterval)
  • If no entries found → forward the record + save the record in the store 
  • If any entries found → do not forward + do not update the store

How to detect late records so as to drop them

The current WindowStore interface does not give information about whether a record is expired or not. put()ing a record does not return whether the record has been actually put or was expired. It does not give as well the max observed stream time info.

...

-----------------------------------

Initial motivation

One example: we might have multiple data sources each reporting its state periodically with a relatively high frequency, their current states should be stored in a database. In case the actual change of the state occurs with a lower frequency than it is reported, in order to reduce the number of writes to the database we might want to filter out duplicated messages using Kafka Streams.

...

Due to 'infinite' nature of KStream, distinct operation should be windowed, similar to windowed joins and aggregations for KStreams.

Initial public interface & examples

In accordance with KStreams DSL Grammar, we introduce the following new elements:

...

The records are considered to be duplicates iff serialized forms of their keys are equal.

Examples

Consider the following example (record times are in seconds):

//three bursts of variously ordered records
4, 5, 6
23, 22, 24
34, 33, 32
//'late arrivals'
7, 22, 35

'Epoch-aligned deduplication' using tumbling windows

.groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(10))).distinct()

...

Note: hopping and sliding windows do not make much sense for distinct() because they produce multiple intersected windows, so that one record can be multiplied instead of deduplication.

SessionWindows work for 'data-aligned deduplication'.

.groupByKey().windowedBy(SessionWindows.with(Duration.ofSeconds(10))).distinct()

...

([key@4000/4000], 4)
([key@23000/23000], 23)
([key@34000/34000], 34)

Initial Rejected public Interfaces

In accordance with KStreams DSL Grammar, we introduce the following new elements:

...

1. KeyValueMapper<K, V, I> idExtractor — extracts a unique identifier from a record by which we de-duplicate input records. If it returns null, the record will not be considered for de-duping and forwarded as-is. If not provided, defaults to (key, value) -> key, which means deduplication based on key of the record. Important assumption: records from different partitions should have different IDs, otherwise same IDs might be not co-partitioned.
2. TimeWindows timeWindows — tumbling or hopping time-based window specification. Required parameter. Only the first message with a given id that falls into a window will be passed downstream.
3. Serde<I> idSerde — serde for unique identifier.
4. boolean isPersistent — whether the WindowStore that stores the unique ids should be persistent or not. In many cases, non-persistent store will be preferrable because of better performance. Downstream consumers must be ready to accept occasional duplicates.

Initial rejected Proposed Changes

1. Add the following method to KStream interface:

...