Join operations (KStream-KStream, KStream-KTable) in Kafka - Commands & Configuration

Introduction

When you have two streams of data flowing in Kafka and want to combine related information from both, join operations help you do that. They let you merge data from two streams or a stream and a table based on matching keys, so you get a richer, combined view.

When you want to combine user click events with user profile updates in real-time.

When you need to enrich order events with product details stored in a table.

When you want to correlate sensor readings from two different devices streaming data.

When you want to join a stream of transactions with a table of account balances to check limits.

When you want to merge two streams of logs from different services based on timestamps.

Config File - stream-join.properties

stream-join.properties

application.id=stream-join-app
bootstrap.servers=localhost:9092
processing.guarantee=exactly_once
cache.max.bytes.buffering=0

# Serdes for keys and values
key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde

# Commit interval
commit.interval.ms=1000

This configuration file sets up the Kafka Streams application with a unique application ID and connects it to the Kafka server at localhost:9092. It ensures exactly-once processing to avoid duplicate results. The key and value serializers/deserializers are set to handle strings. The commit interval controls how often the app saves its progress.

Commands

Create the 'user-clicks' topic where click event data will be streamed.

Terminal

kafka-topics --create --topic user-clicks --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected Output

Created topic user-clicks.

→

--partitions - Number of partitions for parallelism

→

--replication-factor - Number of copies for fault tolerance

Create the 'user-profiles' topic which will be used as a table for user profile data.

Terminal

kafka-topics --create --topic user-profiles --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Expected Output

Created topic user-profiles.

→

--partitions - Number of partitions for parallelism

→

--replication-factor - Number of copies for fault tolerance

Start a producer to send sample click events to the 'user-clicks' topic.

Terminal

kafka-console-producer --topic user-clicks --bootstrap-server localhost:9092

Expected Output

No output (command runs silently)

→

--topic - Topic to send messages to

→

--bootstrap-server - Kafka server address

Start a producer to send user profile updates to the 'user-profiles' topic, which will be used as a table.

Terminal

kafka-console-producer --topic user-profiles --bootstrap-server localhost:9092

Expected Output

No output (command runs silently)

→

--topic - Topic to send messages to

→

--bootstrap-server - Kafka server address

Run the Kafka Streams application that performs KStream-KStream and KStream-KTable joins using the configured properties.

Terminal

java -jar stream-join-app.jar

Expected Output

INFO  Kafka Streams started

INFO  Performing KStream-KStream join on user-clicks and another stream

INFO  Performing KStream-KTable join on user-clicks and user-profiles table

INFO  Streams running...

Consume the output topic 'enriched-clicks' to see the results of the join operations.

Terminal

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

Expected Output

{"userId":"user1","click":"home","profile":"premium"}

{"userId":"user2","click":"search","profile":"basic"}

→

--from-beginning - Read all messages from the start

Key Concept

If you remember nothing else from this pattern, remember: joining streams lets you combine live data flows to enrich or correlate information in real time based on matching keys.

Common Mistakes

Using different key serializers for the two streams or tables being joined

The join will fail because Kafka Streams cannot match keys if they are serialized differently.

Ensure both streams and tables use the same key serde configuration.

Trying to join streams without setting appropriate windowing for KStream-KStream joins

Without windowing, the join will not find matching records because streams are infinite and time-based.

Use a time window (e.g., 5 minutes) to define the join period for KStream-KStream joins.

Not creating the topics before running the streams application

The application will fail to start or produce errors because the input or output topics do not exist.

Create all required topics with correct partitions and replication before starting the app.

Summary

Create Kafka topics for streams and tables before starting the application.

Configure the Kafka Streams app with matching key and value serializers.

Run the app to perform KStream-KStream and KStream-KTable joins to combine data.

Consume the output topic to verify the joined enriched data.

Practice

1. What is the main difference between a KStream-KStream join and a KStream-KTable join in Kafka Streams?

easy

A. KStream-KStream joins require a time window; KStream-KTable joins do not.

B. KStream-KTable joins require a time window; KStream-KStream joins do not.

C. Both joins require a time window to match records.

D. Neither join requires a time window.

Solution

1. Step 1: Understand KStream-KStream join behavior

   KStream-KStream joins combine records from two streams that occur within a specified time window.

2. Step 2: Understand KStream-KTable join behavior

   KStream-KTable joins enrich stream records with the latest matching table record without needing a time window.

3. Final Answer:

   KStream-KStream joins require a time window; KStream-KTable joins do not. -> Option A

4. Quick Check:

   Time window needed only for KStream-KStream joins [OK]

Hint: Remember: only KStream-KStream joins need time windows [OK]

Common Mistakes:

• Confusing which join requires a time window
• Thinking both joins behave the same
• Assuming KStream-KTable join needs time window

2. Which of the following is the correct syntax to perform a KStream-KTable join in Kafka Streams?

easy

A. ktable.join(kstream, (tableValue, streamValue) -> tableValue + streamValue);

B. kstream.join(ktable, JoinWindows.of(Duration.ofMinutes(5)), (v1, v2) -> v1 + v2);

C. kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue);

D. kstream.join(ktable, (v1, v2) -> v1 + v2, JoinWindows.of(Duration.ofMinutes(5)));

Solution

1. Step 1: Identify KStream-KTable join syntax

   KStream-KTable join uses kstream.join(ktable, ValueJoiner) without a time window.

2. Step 2: Check options for correct method signature

   kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue); matches the correct syntax: join without window and with a ValueJoiner lambda.

3. Final Answer:

   kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue); -> Option C

4. Quick Check:

   KStream-KTable join syntax = kstream.join(ktable, (streamValue, tableValue) -> streamValue + tableValue); [OK]

Hint: KStream-KTable join has no time window parameter [OK]

Common Mistakes:

• Adding JoinWindows to KStream-KTable join
• Reversing stream and table parameters
• Using KTable join method on KStream

3. Given the following Kafka Streams code snippet, what will be the output if the input streams have matching keys within the 5-minute window?

var joinedStream = stream1.join(stream2, (v1, v2) -> v1 + v2, JoinWindows.of(Duration.ofMinutes(5)));
joinedStream.foreach((key, value) -> System.out.println(key + ":" + value));

medium

A. Prints combined values for keys appearing in both streams within 5 minutes

B. Prints combined values for all keys in stream1 regardless of stream2

C. Prints only keys from stream2 with no combination

D. Throws a runtime exception due to missing KTable

Solution

1. Step 1: Understand KStream-KStream join with window

   The join combines records from both streams with matching keys occurring within the 5-minute window.

2. Step 2: Analyze output behavior

   For matching keys within the window, the values are combined using the lambda (v1 + v2) and printed.

3. Final Answer:

   Prints combined values for keys appearing in both streams within 5 minutes -> Option A

4. Quick Check:

   KStream-KStream join outputs combined matching keys [OK]

Hint: KStream-KStream join outputs combined values within window [OK]

Common Mistakes:

• Assuming all keys from stream1 print regardless of stream2
• Confusing KStream-KTable join behavior
• Expecting runtime error without KTable

4. Identify the error in the following Kafka Streams join code:

var joined = stream1.join(stream2, (v1, v2) -> v1 + v2);
joined.print();

medium

A. Incorrect lambda syntax for join function

B. Missing JoinWindows parameter for KStream-KStream join

C. Using print() instead of foreach() causes error

D. stream2 should be a KTable, not a KStream

Solution

1. Step 1: Check join method parameters

   KStream-KStream join requires a JoinWindows parameter to specify the time window for matching events.

2. Step 2: Identify missing parameter

   The code omits the JoinWindows argument, causing a compilation error.

3. Final Answer:

   Missing JoinWindows parameter for KStream-KStream join -> Option B

4. Quick Check:

   KStream-KStream join needs JoinWindows [OK]

Hint: KStream-KStream join always needs JoinWindows [OK]

Common Mistakes:

• Forgetting JoinWindows in KStream-KStream join
• Confusing KStream-KTable join syntax
• Assuming print() is invalid method

5. You want to enrich a stream of user clicks (KStream) with the latest user profile data stored in a KTable. Which join operation should you use and why?

hard

A. Use KStream-KStream join without a window to join clicks and profiles.

B. Use KStream-KStream join with a 10-minute window to combine clicks and profiles.

C. Use KTable-KTable join because both are tables.

D. Use KStream-KTable join to enrich clicks with latest profile data without time window.

Solution

1. Step 1: Identify data types and join needs

   User clicks are a stream (KStream), and user profiles are stored as a table (KTable).

2. Step 2: Choose appropriate join type

   KStream-KTable join enriches stream data with the latest table data for matching keys without needing a time window.

3. Final Answer:

   Use KStream-KTable join to enrich clicks with latest profile data without time window. -> Option D

4. Quick Check:

   KStream-KTable join enriches stream with latest table data [OK]

Hint: Enrich stream with table data using KStream-KTable join [OK]

Common Mistakes:

• Using KStream-KStream join unnecessarily
• Adding time window to KStream-KTable join
• Confusing KTable-KTable join for stream enrichment