How to schedule periodic operations in a Kafka Streams application

You'd like to have some periodic functionality execute in your Kafka Streams application. In this tutorial, you'll learn how to use punctuations in Kafka Streams to execute work at regular intervals.

To schedule operations in Kafka Streams, you'll need to use the Processor API. In this example, you will use the KStream.process method which mixes the Processor API into the Kafka Streams DSL.

 final KStream<String, LoginTime> loginTimeStream = builder.stream(INPUT_TOPIC, Consumed.with(Serdes.String(), loginTimeSerde));

        loginTimeStream
                .process(new PunctationProcessorSupplier(), Named.as("max-login-time-transformer"), LOGIN_TIME_STORE)
                      .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));

When you call the KStream.process method, you'll pass in a ProcessorSupplier, which returns your Processor implementation.

It's in the Process.init method where you'll schedule arbitrary actions with the ProcessorContext

 public void init(ProcessorContext<String, Long> context) {
    this.context = context;
    this.context.schedule(Duration.ofSeconds(5), PunctuationType.STREAM_TIME, this::streamTimePunctuator);
    this.context.schedule(Duration.ofSeconds(20), PunctuationType.WALL_CLOCK_TIME, this::wallClockTimePunctuator);
}

Here you're scheduling two actions, one using stream time and another on wall-clock time. With stream time, the event timestamps of the incoming records advance the internal time tracked by Kafka Streams. Wall-clock time is the system time of the machine running the Kafka Streams application. Kafka Streams measures wall-clock during its poll interval so executing punctuation based on wall-clock time is best effort only. Here the code uses method handles instead of concrete Punctuator instances.