eventTime = System.currentTimeMillis();
What is the difference between using the timestamp from the record metadata and using the timestamp within the record payload?
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');
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
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
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
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 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
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.
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';
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
).
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.
Instead of running a local Kafka cluster, you may use Confluent Cloud, a fully-managed Apache Kafka service.
Sign up for Confluent Cloud, a fully-managed Apache Kafka service.
After you log in to Confluent Cloud Console, click on Add cloud environment
and name the environment learn-kafka
. Using a new environment keeps your learning resources separate from your other Confluent Cloud resources.
From the Billing & payment
section in the Menu, apply the promo code CC100KTS
to receive an additional $100 free usage on Confluent Cloud (details).
Click on LEARN and follow the instructions to launch a Kafka cluster and to enable Schema Registry.
Next, from the Confluent Cloud Console, click on Clients
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.