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

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
    }
}

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
    }
}

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
    }
}

Hint

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

Practice

1. What is the main purpose of a punctuator in Kafka Streams?

easy

A. To run code at regular time intervals inside a stream processing application

B. To send messages to Kafka topics

C. To create new Kafka topics automatically

D. To configure Kafka broker settings

Solution

1. Step 1: Understand what punctuators do

   Punctuators are designed to run code periodically during stream processing.

2. Step 2: Identify the correct purpose

   They are not for sending messages or configuring brokers but for timed callbacks.

3. Final Answer:

   To run code at regular time intervals inside a stream processing application -> Option A

4. Quick Check:

   Punctuator purpose = run code regularly [OK]

Hint: Punctuators trigger code on time intervals, not message sending [OK]

Common Mistakes:

• Confusing punctuators with message producers
• Thinking punctuators create topics
• Assuming punctuators configure Kafka brokers

2. Which of the following is the correct way to schedule a punctuator in Kafka Streams?

easy

A. context.schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, callback)

B. context.schedule(10, TimeUnit.SECONDS, callback)

C. context.setTimer(10, callback)

D. context.runEvery(10, callback)

Solution

1. Step 1: Recall the correct method signature

   The Kafka Streams API uses context.schedule() with Duration, PunctuationType, and callback.

2. Step 2: Match the correct syntax

   context.schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, callback) matches the correct method call with Duration and PunctuationType.

3. Final Answer:

   context.schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, callback) -> Option A

4. Quick Check:

   Correct schedule syntax = context.schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, callback) [OK]

Hint: Use context.schedule with Duration and PunctuationType [OK]

Common Mistakes:

• Using incorrect method names like setTimer or runEvery
• Passing integers instead of Duration objects
• Omitting PunctuationType argument

3. Given the following code snippet in a Kafka Streams processor:

context.schedule(Duration.ofMinutes(1), PunctuationType.STREAM_TIME, timestamp -> {
    System.out.println("Punctuator triggered at " + timestamp);
});

What will be printed when the punctuator triggers?

medium

A. Punctuator triggered at current system time in milliseconds

B. Punctuator triggered at the stream time (event time) in milliseconds

C. Punctuator triggered at a fixed string 'timestamp'

D. No output because punctuator is not scheduled correctly

Solution

1. Step 1: Identify the PunctuationType used

   The code uses PunctuationType.STREAM_TIME, which triggers based on event time in the stream.

2. Step 2: Understand the timestamp parameter

   The timestamp passed is the stream time in milliseconds, not system time.

3. Final Answer:

   Punctuator triggered at the stream time (event time) in milliseconds -> Option B

4. Quick Check:

   PunctuationType.STREAM_TIME = stream event time [OK]

Hint: STREAM_TIME triggers use event time, not system clock [OK]

Common Mistakes:

• Confusing STREAM_TIME with WALL_CLOCK_TIME
• Expecting system time instead of stream time
• Assuming no output due to scheduling

4. Identify the error in this Kafka Streams punctuator scheduling code:

context.schedule(5000, PunctuationType.WALL_CLOCK_TIME, timestamp -> {
    System.out.println("Triggered");
});

medium

A. context.schedule cannot be called inside a processor

B. PunctuationType should be STREAM_TIME, not WALL_CLOCK_TIME

C. Callback function must return a value

D. Duration object is missing; integer milliseconds used instead

Solution

1. Step 1: Check the schedule method parameters

   The schedule method expects a Duration object, not an integer.

2. Step 2: Identify the incorrect argument type

   Passing 5000 as int instead of Duration.ofMillis(5000) causes a syntax error.

3. Final Answer:

   Duration object is missing; integer milliseconds used instead -> Option D

4. Quick Check:

   Use Duration, not int, for schedule interval [OK]

Hint: Always use Duration.ofMillis or similar, not raw integers [OK]

Common Mistakes:

• Passing raw integers instead of Duration
• Confusing PunctuationType values
• Thinking callback must return a value

5. You want to emit aggregated results every 30 seconds using a punctuator in Kafka Streams. Which of the following code snippets correctly schedules this punctuator and ensures it triggers based on wall-clock time?

hard

A. context.schedule(30, PunctuationType.WALL_CLOCK_TIME, timestamp -> emitResults())

B. context.schedule(Duration.ofSeconds(30), PunctuationType.STREAM_TIME, timestamp -> emitResults())

C. context.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME, timestamp -> emitResults())

D. context.schedule(Duration.ofMinutes(1), PunctuationType.WALL_CLOCK_TIME, timestamp -> emitResults())

Solution

1. Step 1: Determine the correct interval and time type

   The requirement is every 30 seconds and based on wall-clock time.

2. Step 2: Match the correct schedule call

   context.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME, timestamp -> emitResults()) uses Duration.ofSeconds(30) and PunctuationType.WALL_CLOCK_TIME, which is correct.

3. Step 3: Check other options for errors

   context.schedule(Duration.ofSeconds(30), PunctuationType.STREAM_TIME, timestamp -> emitResults()) uses STREAM_TIME, not wall-clock; C uses int instead of Duration; D uses 1 minute instead of 30 seconds.

4. Final Answer:

   context.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME, timestamp -> emitResults()) -> Option C

5. Quick Check:

   Wall-clock 30s interval = context.schedule(Duration.ofSeconds(30), PunctuationType.WALL_CLOCK_TIME, timestamp -> emitResults()) [OK]

Hint: Use Duration.ofSeconds(30) with WALL_CLOCK_TIME for 30-second triggers [OK]

Common Mistakes:

• Using STREAM_TIME instead of WALL_CLOCK_TIME
• Passing raw integers instead of Duration
• Setting wrong time interval