Punctuators for time-based triggers in Kafka - Mini Project: Build & Apply

Punctuators for time-based triggers

📖 Scenario: You are building a Kafka Streams application that processes user activity events. You want to perform an action at regular time intervals, such as logging or updating a state, using punctuators.

🎯 Goal: Learn how to set up a punctuator in Kafka Streams that triggers an action every 10 seconds.

📋 What You'll Learn

💡 Why This Matters

🌍 Real World

Time-based punctuators are useful for periodic tasks like metrics reporting, state cleanup, or triggering alerts in streaming applications.

💼 Career

Understanding punctuators is important for Kafka Streams developers to implement time-driven logic efficiently.

1

Create a Kafka Streams topology with a source processor

Create a StreamsBuilder object called builder and define a source stream named user-activity assigned to a variable called sourceStream.

Kafka

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

// Your code here

Need a hint?

Use StreamsBuilder to create the topology and call builder.stream with the topic name.

2

Add a punctuator that triggers every 10 seconds

Create a ProcessorSupplier called processorSupplier that returns a Processor which schedules a punctuator to run every 10 seconds (10000 milliseconds).

Kafka

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.Cancellable;
import org.apache.kafka.streams.processor.ProcessorContext;

public class PunctuatorExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream sourceStream = builder.stream("user-activity");

        // Add a ProcessorSupplier called processorSupplier that creates a Processor which schedules a punctuator every 10 seconds
        // Your code here
    }
}

Need a hint?

Use context.schedule with 10000 milliseconds and PunctuationType.WALL_CLOCK_TIME inside the Processor's init method.

3

Attach the processor to the topology

Use sourceStream.process(processorSupplier) to attach the processor with the punctuator to the stream.

Kafka

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.Cancellable;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

public class PunctuatorExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream sourceStream = builder.stream("user-activity");

        ProcessorSupplier processorSupplier = () -> new Processor() {
            private ProcessorContext context;
            private Cancellable cancellable;

            @Override
            public void init(ProcessorContext context) {
                this.context = context;
                this.cancellable = context.schedule(10000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
                    System.out.println("Punctuator triggered at " + timestamp);
                });
            }

            @Override
            public void process(String key, String value) {
                // No processing needed here
            }

            @Override
            public void close() {
                if (cancellable != null) {
                    cancellable.cancel();
                }
            }
        };

        // Attach the processorSupplier to sourceStream
        // Your code here
    }
}

Need a hint?

Call process on sourceStream with processorSupplier as argument.

4

Start the Kafka Streams application and observe the output

Create a KafkaStreams object called streams using builder.build() and a Properties object called props with application ID punctuator-app and bootstrap servers localhost:9092. Start the streams and print a message when the punctuator triggers.

Kafka

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.processor.Punctuator;
import org.apache.kafka.streams.processor.Cancellable;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;
import org.apache.kafka.streams.KafkaStreams;
import java.util.Properties;

public class PunctuatorExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream sourceStream = builder.stream("user-activity");

        ProcessorSupplier processorSupplier = () -> new Processor() {
            private ProcessorContext context;
            private Cancellable cancellable;

            @Override
            public void init(ProcessorContext context) {
                this.context = context;
                this.cancellable = context.schedule(10000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
                    System.out.println("Punctuator triggered at " + timestamp);
                });
            }

            @Override
            public void process(String key, String value) {
                // No processing needed here
            }

            @Override
            public void close() {
                if (cancellable != null) {
                    cancellable.cancel();
                }
            }
        };

        sourceStream.process(processorSupplier);

        // Create Properties props with application.id and bootstrap.servers
        // Create KafkaStreams streams with builder.build() and props
        // Start streams
        // Your code here
    }
}

Need a hint?

Create Properties, set application.id and bootstrap.servers, then create KafkaStreams and call start().