Event-time semantics

Question:

What is the difference between using the timestamp from the record metadata and using the timestamp within the record payload?

Edit this page

Example use case:

By default, time-based aggregations in Kafka Streams and ksqlDB (tumbling windows, hopping windows, etc.) operate on the timestamp in the record metadata, which could be either 'CreateTime' (producer system time) or 'LogAppendTime' (broker system time), depending on the message.timestamp.type configuration value. 'CreateTime' helps with event-time semantics, but there may be use cases where the desired event time is a timestamp embedded inside the record payload. This tutorial shows you how to use the timestamp either from the record metadata or from a field in the record payload.

Code example:





Short Answer

Every record in ksqlDB has a system-column called ROWTIME that tracks the timestamp of the event. By default, ROWTIME is inherited from the timestamp in the underlying Kafka record metadata. To use the timestamp from a field in the record payload instead, configure the TIMESTAMP option when you create the stream:

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

Try it

1
Initialize the project

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

2
Get Confluent Platform

Next, create the following docker-compose.yml file to obtain Confluent Platform:

---
version: '2'

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

  broker:
    image: confluentinc/cp-kafka:6.1.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:6.1.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.17.0
    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.17.0
    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

3
Configure the project

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 "com.commercehub.gradle.plugin:gradle-avro-plugin:0.21.0"
        classpath "com.github.jengelman.gradle.plugins:shadow:4.0.2"
    }
}

plugins {
    id "java"
    id "com.google.cloud.tools.jib" version "2.6.0"
    id "idea"
    id "eclipse"
}

sourceCompatibility = "1.8"
targetCompatibility = "1.8"
version = "0.0.1"

repositories {
    mavenCentral()

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

apply plugin: "com.commercehub.gradle.plugin.avro"
apply plugin: "com.github.johnrengelman.shadow"

dependencies {
    implementation "org.apache.avro:avro:1.10.0"
    implementation "org.slf4j:slf4j-simple:1.7.30"
    implementation "org.apache.kafka:kafka-streams:2.7.0"
    implementation "io.confluent:kafka-streams-avro-serde:6.1.0"
    testImplementation "org.apache.kafka:kafka-streams-test-utils:2.7.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

4
Create a schema for the events

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

5
Produce Kafka data with event time in the payload

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.

6
Write the program interactively using the CLI

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';

7
Processing based on log time

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).

8
Processing based on event time

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.

Deploy on Confluent Cloud

1
Run your app to Confluent Cloud

Instead of running a local Kafka cluster, you may use Confluent Cloud, a fully-managed Apache Kafka service.

First, create your Kafka cluster in Confluent Cloud. Use the promo code CC100KTS to receive an additional $100 free usage (details).

Next, from the Confluent Cloud UI, click on Tools & client config to get the cluster-specific configurations, e.g. Kafka cluster bootstrap servers and credentials, Confluent Cloud Schema Registry and credentials, etc., and set the appropriate parameters in your client application.

Now you’re all set to run your streaming application locally, backed by a Kafka cluster fully managed by Confluent Cloud.