Get Started Free
Tutorial

How to achieve temporal join accuracy in Kafka Streams with versioned KTables

How to achieve temporal join accuracy in Kafka Streams with versioned KTables

Proper handling of time in Kafka Stream stream-table joins has historically been difficult to achieve. It used to be when Kafka Streams executes a stream-table join the stream side event would join the latest available record with the same key on the table side. But, sometimes it's important for the stream event to match up with a table record by timestamp as well as key. Consider a stream of stock transactions and a table of stock prices -- it's essential the transaction joins with the stock price at the time of the transaction, not the latest price. A versioned state store tracks multiple record versions for the same key, rather than the single latest record per key, as is the case for standard non-versioned stores.

The key to versioned state stores is to use a VersionedKeyValueStore when creating a KTable:

    final VersionedBytesStoreSupplier versionedStoreSupplier =
              Stores.persistentVersionedKeyValueStore("versioned-ktable-store",
                                                       Duration.ofMinutes(10));


    final KTable<String, String> tableInput = builder.table(tableInputTopic,
                Materialized.<String, String>as(versionedStoreSupplier)
                        .withKeySerde(stringSerde)
                        .withValueSerde(stringSerde));

Assuming you have a versioned KTable and a KStream with out-of-order records to join, the join will be temporally correct since each stream record with be joined with a table record aligned by timestamp instead of simply using the latest record for the key.

The following steps use Confluent Cloud. To run the tutorial locally with Docker, skip to the Docker instructions section at the bottom.

Prerequisites

  • A Confluent Cloud account
  • The Confluent CLI installed on your machine
  • Apache Kafka or Confluent Platform (both include the Kafka Streams application reset tool)
  • Clone the confluentinc/tutorials repository and navigate into its top-level directory:
    git clone git@github.com:confluentinc/tutorials.git
    cd tutorials

Create Confluent Cloud resources

Login to your Confluent Cloud account:

confluent login --prompt --save

Install a CLI plugin that will streamline the creation of resources in Confluent Cloud:

confluent plugin install confluent-quickstart

Run the plugin from the top-level directory of the tutorials repository to create the Confluent Cloud resources needed for this tutorial. Note that you may specify a different cloud provider (gcp or azure) or region. You can find supported regions in a given cloud provider by running confluent kafka region list --cloud <CLOUD>.

confluent quickstart \
  --environment-name kafka-streams-versioned-ktable-join-env \
  --kafka-cluster-name kafka-streams-versioned-ktable-join-cluster \
  --create-kafka-key \
  --kafka-java-properties-file ./versioned-ktables/kstreams/src/main/resources/cloud.properties

The plugin should complete in under a minute.

Create topics

Create the input and output topics for the application:

confluent kafka topic create stream-input-topic
confluent kafka topic create table-input-topic
confluent kafka topic create output-topic

Start a console producer:

confluent kafka topic produce table-input-topic --parse-key --delimiter :

Enter a few strings representing the second half of common phrases:

one:jelly
two:cheese
three:crackers
four:biscuits
five:cream

Start a console producer:

confluent kafka topic produce stream-input-topic --parse-key --delimiter :

Enter a few strings representing the correct first half of the common phrases:

one:peanut butter and
two:ham and
three:cheese and
four:tea and
five:coffee with

Enter Ctrl+C to exit the console producer.

Finally, start a console producer:

confluent kafka topic produce table-input-topic --parse-key --delimiter :

Enter a few strings representing the incorrect second half of common phrases:

one:sardines
two:an old tire
three:fish eyes
four:moldy bread
five:lots of salt

Enter Ctrl+C to exit the console producer.

Compile and run the application

Compile the application from the top-level tutorials repository directory:

./gradlew versioned-ktables:kstreams:shadowJar

Navigate into the application's home directory:

cd versioned-ktables/kstreams

Run the application, passing the Kafka client configuration file generated when you created Confluent Cloud resources:

java -cp ./build/libs/versioned-ktables-standalone.jar \
    io.confluent.developer.VersionedKTableExample \
    ./src/main/resources/cloud.properties

Validate that you see the correct phrases in the output-topic topic.

confluent kafka topic consume output-topic -b

You should see:

peanut butter and jelly
ham and cheese
cheese and crackers
tea and biscuits
coffee with cream

Clean up

When you are finished, delete the kafka-streams-versioned-ktable-join-env environment by first getting the environment ID of the form env-123456 corresponding to it:

confluent environment list

Delete the environment, including all resources created for this tutorial:

confluent environment delete <ENVIRONMENT ID>
Docker instructions

Prerequisites

  • Docker running via Docker Desktop or Docker Engine
  • Docker Compose. Ensure that the command docker compose version succeeds.
  • Clone the confluentinc/tutorials repository and navigate into its top-level directory:
    git clone git@github.com:confluentinc/tutorials.git
    cd tutorials

Start Kafka in Docker

Start Kafka with the following command run from the top-level tutorials repository directory:

docker compose -f ./docker/docker-compose-kafka.yml up -d

Create topics

Open a shell in the broker container:

docker exec -it broker /bin/bash

Create the input and output topics for the application:

kafka-topics --bootstrap-server localhost:9092 --create --topic stream-input-topic
kafka-topics --bootstrap-server localhost:9092 --create --topic table-input-topic
kafka-topics --bootstrap-server localhost:9092 --create --topic output-topic

Start a console producer:

kafka-console-producer --bootstrap-server localhost:9092 --topic table-input-topic \
    --property "parse.key=true" --property "key.separator=:"

Enter a few strings representing the second half of common phrases:

one:jelly
two:cheese
three:crackers
four:biscuits
five:cream

Start a console producer:

kafka-console-producer --bootstrap-server localhost:9092 --topic stream-input-topic \
    --property "parse.key=true" --property "key.separator=:"

Enter a few strings representing the correct first half of the common phrases:

one:peanut butter and
two:ham and
three:cheese and
four:tea and
five:coffee with

Enter Ctrl+C to exit the console producer.

Finally, start a console producer:

kafka-console-producer --bootstrap-server localhost:9092 --topic table-input-topic \
    --property "parse.key=true" --property "key.separator=:"

Enter a few strings representing the incorrect second half of common phrases:

one:sardines
two:an old tire
three:fish eyes
four:moldy bread
five:lots of salt

Enter Ctrl+C to exit the console producer.

Compile and run the application

On your local machine, compile the app:

./gradlew versioned-ktables:kstreams:shadowJar

Navigate into the application's home directory:

cd versioned-ktables/kstreams

Run the application, passing the local.properties Kafka client configuration file that points to the broker's bootstrap servers endpoint at localhost:9092:

java -cp ./build/libs/versioned-ktables-standalone.jar \
    io.confluent.developer.VersionedKTableExample \
    ./src/main/resources/local.properties

Validate that you see the correct phrases in the output-topic topic. In the broker container shell:

kafka-console-consumer --bootstrap-server localhost:9092 --topic output-topic --from-beginning

You should see the correct phrases:

peanut butter and jelly
ham and cheese
cheese and crackers
tea and biscuits
coffee with cream

Clean up

From your local machine, stop the broker container:

docker compose -f ./docker/docker-compose-kafka.yml down
Do you have questions or comments? Join us in the #confluent-developer community Slack channel to engage in discussions with the creators of this content.