Debounce kafka events

At present there's no easy way to debounce events.

The problem, in short, is that Kafka doesn't act based on 'wall clock time'. Processing is generally triggered by incoming events (and the data contained therein), not by arbitrary triggers, like system time.

I'll cover why Suppressed and SessionWindows don't work, the proposed solution in KIP-242, and an untested workaround.

Suppressed

Suppressed has a untilTimeLimit() function, but it isn't suitable for debouncing.

If another record for the same key arrives in the mean time, it replaces the first record in the buffer but does not re-start the timer.

SessionWindows

I thought that using SessionWindows.ofInactivityGapAndGrace() might work.

First I grouped, windowed, aggregated, and suppressed the input KStream:

  val windowedData: KTable<Windowed<Key>, Value> = 
    inputTopicKStream
      .groupByKey()
      .windowedBy(
        SessionWindows.ofInactivityGapAndGrace(
          WINDOW_INACTIVITY_DURATION,
          WINDOW_INACTIVITY_DURATION,
        )
      )
      .aggregate(...)
      .suppress(
        Suppressed.untilWindowCloses(
          Suppressed.BufferConfig.unbounded()
        )
      )

Then I aggregated the windows, so I could have a final state

  windowedData
      .groupBy(...)
      .reduce(
        /* adder */
        { a, b -> a + b },
        /* subtractor */
        { a, a -> a - a },
      )

However the problem is that SessionWindows will not close without additional records coming up. Kafka will not independently close the window - it requires additional records to realise that the window can be closed, and that suppress() can forward a new record.

This is noted in Confluent's blog https://www.confluent.io/de-de/blog/kafka-streams-take-on-watermarks-and-triggers/

[I]f you stop getting new records wall-clock time will continue to advance, but stream time will freeze. Wall-clock time advances because that little quartz watch in your computer keeps ticking away, but stream time only advances when you get new records. With no new records, stream time is frozen.

KIP-424

KIP-424 proposed an improvement that would allow Suppress to act as a debouncer, but there's been no progress in a couple of years.

Workaround

Andrey Bratus provided a simple workaround in the JIRA ticket for KIP-424, KAFKA-7748. I tried it but it didn't compile - I think the Kafka API has evolved since the workaround was posted. I've updated the code, but I haven't tested it.

import java.time.Duration;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.state.TimestampedKeyValueStore;
import org.apache.kafka.streams.state.ValueAndTimestamp;

/**
 * THIS PROCESSOR IS UNTESTED
 * <br>
 * This processor mirrors the source, but waits for an inactivity gap before forwarding records.
 * <br>
 * The suppression is key based. Newer values will replace previous values, and reset the inactivity
 * gap.
 */
public class SuppressProcessor<K, V> implements Processor<K, V, K, V> {

  private final String storeName;
  private final Duration debounceCheckInterval;
  private final long suppressTimeoutMillis;

  private TimestampedKeyValueStore<K, V> stateStore;
  private ProcessorContext<K, V> context;

  /**
   * @param storeName             The name of the {@link TimestampedKeyValueStore} which will hold
   *                              records while they are being debounced.
   * @param suppressTimeout       The duration of inactivity before records will be forwarded.
   * @param debounceCheckInterval How regularly all records will be checked to see if they are
   *                              eligible to be forwarded. The interval should be shorter than
   *                              {@code suppressTimeout}.
   */
  public SuppressProcessor(
      String storeName,
      Duration suppressTimeout,
      Duration debounceCheckInterval
  ) {
    this.storeName = storeName;
    this.suppressTimeoutMillis = suppressTimeout.toMillis();
    this.debounceCheckInterval = debounceCheckInterval;
  }

  @Override
  public void init(ProcessorContext<K, V> context) {
    this.context = context;

    stateStore = context.getStateStore(storeName);

    context.schedule(debounceCheckInterval, PunctuationType.WALL_CLOCK_TIME, this::punctuate);
  }

  @Override
  public void process(Record<K, V> record) {

    final var key = record.key();
    final var value = record.value();

    final var storedRecord = stateStore.get(key);

    final var isNewRecord = storedRecord == null;

    final var timestamp = isNewRecord ? System.currentTimeMillis() : storedRecord.timestamp();

    stateStore.put(key, ValueAndTimestamp.make(value, timestamp));
  }

  private void punctuate(long timestamp) {
    try (var iterator = stateStore.all()) {
      while (iterator.hasNext()) {
        KeyValue<K, ValueAndTimestamp<V>> storedRecord = iterator.next();
        if (timestamp - storedRecord.value.timestamp() > suppressTimeoutMillis) {

          final var record = new Record<>(
              storedRecord.key,
              storedRecord.value.value(),
              storedRecord.value.timestamp()
          );

          context.forward(record);
          stateStore.delete(storedRecord.key);
        }
      }
    }
  }
}

If you are using a Kafka Streams app, you could try to use suppress

It is designed for WindowedKStream and KTable to "hold back" an update and very useful for rate limiting or notification at the end of a window.

There is a quite useful explanation on https://www.confluent.de/blog/kafka-streams-take-on-watermarks-and-triggers/