Event-time semantics using ksqlDB

Prerequisites

1

This tutorial installs Confluent Platform using Docker. Before proceeding:

• Install Docker Desktop (version 4.0.0 or later) or Docker Engine (version 19.03.0 or later) if you don’t already have it
• Install the Docker Compose plugin if you don’t already have it. This isn’t necessary if you have Docker Desktop since it includes Docker Compose.
• Start Docker if it’s not already running, either by starting Docker Desktop or, if you manage Docker Engine with systemd, via systemctl
• Verify that Docker is set up properly by ensuring no errors are output when you run docker info and docker compose version on the command line

Initialize the project

2

To get started, make a new directory anywhere you’d like for this project:

mkdir time-concepts && cd time-concepts

Then make the following directories to set up its structure:

mkdir src

Get Confluent Platform

3

Next, create the following docker-compose.yml file to obtain Confluent Platform (for Kafka in the cloud, see Confluent Cloud):

---
version: '2'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka:7.3.0
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.0
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:9092'

  ksqldb-server:
    image: confluentinc/ksqldb-server:0.28.2
    hostname: ksqldb-server
    container_name: ksqldb-server
    depends_on:
      - broker
      - schema-registry
    ports:
      - "8088:8088"
    environment:
      KSQL_CONFIG_DIR: "/etc/ksqldb"
      KSQL_LOG4J_OPTS: "-Dlog4j.configuration=file:/etc/ksqldb/log4j.properties"
      KSQL_BOOTSTRAP_SERVERS: "broker:9092"
      KSQL_HOST_NAME: ksqldb-server
      KSQL_LISTENERS: "http://0.0.0.0:8088"
      KSQL_CACHE_MAX_BYTES_BUFFERING: 0
      KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      KSQL_KSQL_STREAMS_AUTO_OFFSET_RESET: "earliest"

  ksqldb-cli:
    image: confluentinc/ksqldb-cli:0.28.2
    container_name: ksqldb-cli
    depends_on:
      - broker
      - ksqldb-server
    entrypoint: /bin/sh
    environment:
      KSQL_CONFIG_DIR: "/etc/ksqldb"
    tty: true
    volumes:
      - ./src:/opt/app/src

And launch it by running:

docker compose up -d

Configure the project

4

This example uses a Kafka Producer application to write events to Kafka with an artificial delay between the simulated event time and producing the event to Kafka, to exaggerate the difference between these times. Because this example runs a Kafka application, the next few steps will pull in requirements to build out your application.

Create the following Gradle build file for the project, named build.gradle:

buildscript {
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath "gradle.plugin.com.github.jengelman.gradle.plugins:shadow:7.0.0"
    }
}

plugins {
    id "java"
    id "com.google.cloud.tools.jib" version "2.6.0"
    id "idea"
    id "eclipse"
    id "com.github.davidmc24.gradle.plugin.avro" version "1.5.0"
}

sourceCompatibility = JavaVersion.VERSION_17
targetCompatibility = JavaVersion.VERSION_17
version = "0.0.1"

repositories {
    mavenCentral()

    maven {
        url "https://packages.confluent.io/maven"
    }
}

apply plugin: "com.github.johnrengelman.shadow"

dependencies {
    implementation "org.apache.avro:avro:1.11.0"
    implementation "org.slf4j:slf4j-simple:1.7.30"
    implementation 'org.apache.kafka:kafka-streams:3.4.0'
    implementation ('org.apache.kafka:kafka-clients') {
       version {
           strictly '3.4.0'
        }
      }
    implementation "io.confluent:kafka-streams-avro-serde:7.3.0"
    testImplementation "org.apache.kafka:kafka-streams-test-utils:3.4.0"
    testImplementation "junit:junit:4.13.1"
    testImplementation 'org.hamcrest:hamcrest:2.2'
}

test {
    testLogging {
        outputs.upToDateWhen { false }
        showStandardStreams = true
        exceptionFormat = "full"
    }
}

jar {
  manifest {
    attributes(
      "Class-Path": configurations.compileClasspath.collect { it.getName() }.join(" "),
      "Main-Class": "io.confluent.developer.KafkaProducerDevice"
    )
  }
}

shadowJar {
    archiveBaseName = "kafka-producer-device-standalone"
    archiveClassifier = ''
}

Run the following command to obtain the Gradle wrapper:

gradle wrapper

Next, create a directory for configuration data:

mkdir configuration

Then create a development file at configuration/dev.properties:

application.id=time-concepts
bootstrap.servers=127.0.0.1:29092
schema.registry.url=http://127.0.0.1:8081

output.topic.name=deviceEvents
output.topic.partitions=1
output.topic.replication.factor=1

Create a schema for the events

5

Create a directory for the schemas that represent the events in the stream:

mkdir -p src/main/avro

Then create the following Avro schema file at src/main/avro/DeviceEvent.avsc for the event. This schema has two fields, one of which is called eventTime that represents the event time.

{
  "namespace": "io.confluent.developer.avro",
  "type": "record",
  "name": "DeviceEvent",
  "fields": [
    {"name": "temperature", "type": "long"},
    {"name": "eventTime", "type": "long"}
  ]
}

Because this Avro schema is used in the Java code, it needs to compile it. Run the following:

./gradlew build

Produce Kafka data with event time in the payload

6

Create a directory for the Java files in this project:

mkdir -p src/main/java/io/confluent/developer

Achieving event-time semantics typically requires embedding timestamps into the data record at the time it is produced. Write a Kafka Producer application that generates simulated device events and embeds a timestamp into the payload of every message. The timestamp is written in an arbitrary field, in this case called eventTime, whose value represents the time at which the event occurred at the source.

eventTime = System.currentTimeMillis();

Create the full application file at src/main/java/io/confluent/developer/KafkaProducerDevice.java.

package io.confluent.developer;


import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.common.serialization.LongSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;

import java.io.FileInputStream;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Collection;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

import io.confluent.developer.avro.DeviceEvent;

public class KafkaProducerDevice {

    private final Producer<Long, DeviceEvent> producer;
    final String outTopic;

    public KafkaProducerDevice(final Producer<Long, DeviceEvent> producer,
                                    final String topic) {
        this.producer = producer;
        outTopic = topic;
    }

    public void shutdown() {
        producer.close();
    }

    public static Properties loadProperties(String fileName) throws IOException {
        final Properties envProps = new Properties();
        final FileInputStream input = new FileInputStream(fileName);
        envProps.load(input);
        input.close();

        return envProps;
    }

    public static void main(String[] args) throws Exception {
        if (args.length < 1) {
            throw new IllegalArgumentException(
                    "This program takes one argument: the path to an environment configuration file");
        }

        final Properties props = KafkaProducerDevice.loadProperties(args[0]);

        props.put(ProducerConfig.ACKS_CONFIG, "all");

        props.put(ProducerConfig.CLIENT_ID_CONFIG, "myEventApp");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);

        final String topic = props.getProperty("output.topic.name");
        final Producer<Long, DeviceEvent> producer = new KafkaProducer<Long, DeviceEvent>(props);
        final KafkaProducerDevice producerApp = new KafkaProducerDevice(producer, topic);

        final Long deviceId = 1L;
        Long temperature = 100L;
        Long eventTime;

        int count = 0;

        while(count < 10) {

            eventTime = System.currentTimeMillis();
            // Inject artificial delay before record is produced to Kafka
            // to force differing timestamps in payload and metadata
            Thread.sleep(1005);

            DeviceEvent record = new DeviceEvent(temperature, eventTime);
            final ProducerRecord<Long, DeviceEvent> producerRecord = new ProducerRecord<>(topic, deviceId, record);
            producer.send(producerRecord,
                (recordMetadata, e) -> {
                    if(e != null) {
                       e.printStackTrace();
                    } else {
                      System.out.println("Record written to topic " + recordMetadata.topic() + ": payload eventTime " + record.getEventTime() + ", Kafka timestamp " + recordMetadata.timestamp());
                    }
                  }
                );

            count++;
            temperature++;

        }

        producerApp.shutdown();

    }
}

In your terminal, run:

./gradlew shadowJar

Now that you have an uberjar for the KafkaProducerDevice application, you can launch it locally.

java -jar build/libs/kafka-producer-device-standalone-0.0.1.jar configuration/dev.properties

After you run the previous command, the application will write some messages to Kafka and you should something like this on the console:

Record written to topic deviceEvents: payload eventTime 1606785578301, Kafka timestamp 1606785579368
Record written to topic deviceEvents: payload eventTime 1606785579482, Kafka timestamp 1606785580492
Record written to topic deviceEvents: payload eventTime 1606785580492, Kafka timestamp 1606785581501
Record written to topic deviceEvents: payload eventTime 1606785581501, Kafka timestamp 1606785582511
Record written to topic deviceEvents: payload eventTime 1606785582512, Kafka timestamp 1606785583521
Record written to topic deviceEvents: payload eventTime 1606785583521, Kafka timestamp 1606785584528
Record written to topic deviceEvents: payload eventTime 1606785584528, Kafka timestamp 1606785585533
Record written to topic deviceEvents: payload eventTime 1606785585534, Kafka timestamp 1606785586544
Record written to topic deviceEvents: payload eventTime 1606785586544, Kafka timestamp 1606785587551
Record written to topic deviceEvents: payload eventTime 1606785587552, Kafka timestamp 1606785588562

Note that the payload eventTime value is not the same as the Kafka timestamp value—this is working as expected. It demonstrates how the end system may intentionally set an event time in the payload, and it will differ from the Kafka record metadata timestamp.

Write the program interactively using the CLI

7

The best way to interact with ksqlDB when you’re learning how things work is with the ksqlDB CLI. Fire it up as follows:

docker exec -it ksqldb-cli ksql http://ksqldb-server:8088

Before we get too far, let’s set the auto.offset.reset configuration parameter to earliest. This means all new ksqlDB queries will automatically compute their results from the beginning of a stream, rather than the end. This isn’t always what you’ll want to do in production, but it makes query results much easier to see in examples like this.

SET 'auto.offset.reset' = 'earliest';

Processing based on log time

8

Run the following ksqlDB command to create a stream of events from the underlying Kafka topic deviceEvents (which was the topic written to by the Kafka producer application above).

CREATE STREAM TEMPERATURE_READINGS_LOGTIME
    WITH (KAFKA_TOPIC='deviceEvents',
          VALUE_FORMAT='avro');

Let’s inspect the events in this newly created TEMPERATURE_READINGS_LOGTIME stream by running a SELECT statement with an EMIT CHANGES clause, limited to 10. It shows the payload fields TEMPERATURE and EVENTTIME, plus the ROWTIME which is a system-column in ksqlDB that is used for time-based aggregations.

SELECT *, ROWTIME
FROM TEMPERATURE_READINGS_LOGTIME
EMIT CHANGES
LIMIT 10;

This should yield the following output:

+--------------------------------+--------------------------------+--------------------------------+
|TEMPERATURE                     |EVENTTIME                       |ROWTIME                         |
+--------------------------------+--------------------------------+--------------------------------+
|100                             |1606785578301                   |1606785579368                   |
|101                             |1606785579482                   |1606785580492                   |
|102                             |1606785580492                   |1606785581501                   |
|103                             |1606785581501                   |1606785582511                   |
|104                             |1606785582512                   |1606785583521                   |
|105                             |1606785583521                   |1606785584528                   |
|106                             |1606785584528                   |1606785585533                   |
|107                             |1606785585534                   |1606785586544                   |
|108                             |1606785586544                   |1606785587551                   |
|109                             |1606785587552                   |1606785588562                   |
Limit Reached
Query terminated

Notice that for each row:

The EVENTTIME value in ksqlDB corresponds exactly to the payload eventTime, a field within the record payload
The ROWTIME value in ksqlDB corresponds exactly to the Kafka timestamp printed by the producer’s callback, corresponding to the record metadata timestamp

Any time-based aggregations on this stream is based on ROWTIME, consequently this results in processing based on the timestamp in the Kafka record metadata (either CreateTime or LogAppendTime).

Processing based on event time

9

Now create a new stream, but force ksqlDB to use the eventTime field in the payload of each record as the timestamp, by setting the TIMESTAMP parameter. (For details on managing timestamps in ksqlDB, read more at How to use a custom timestamp column).

CREATE STREAM TEMPERATURE_READINGS_EVENTTIME
    WITH (KAFKA_TOPIC='deviceEvents',
          VALUE_FORMAT='avro',
          TIMESTAMP='eventTime');

Inspect the events in this newly created TEMPERATURE_READINGS_EVENTTIME stream by running:

SELECT *, ROWTIME
FROM TEMPERATURE_READINGS_EVENTTIME
EMIT CHANGES
LIMIT 10;

This should yield the following output:

+--------------------------------+--------------------------------+--------------------------------+
|TEMPERATURE                     |EVENTTIME                       |ROWTIME                         |
+--------------------------------+--------------------------------+--------------------------------+
|100                             |1606785578301                   |1606785578301                   |
|101                             |1606785579482                   |1606785579482                   |
|102                             |1606785580492                   |1606785580492                   |
|103                             |1606785581501                   |1606785581501                   |
|104                             |1606785582512                   |1606785582512                   |
|105                             |1606785583521                   |1606785583521                   |
|106                             |1606785584528                   |1606785584528                   |
|107                             |1606785585534                   |1606785585534                   |
|108                             |1606785586544                   |1606785586544                   |
|109                             |1606785587552                   |1606785587552                   |
Limit Reached
Query terminated

Notice that for each row:

The ROWTIME value in ksqlDB corresponds exactly to the EVENTTIME, which is the payload eventTime, a field within the record payload.

Any time-based aggregations on this stream is based on ROWTIME, consequently this results in processing based on event time.

Event-time semantics

Question:

Example use case:

Hands-on code example:

Short Answer

Run it