Join operations (KStream-KStream, KStream-KTable) in Kafka - Step-by-Step Execution

Process Flow - Join operations (KStream-KStream, KStream-KTable)

Start with two streams or stream+table

↓

Match keys from both sides

↓

Apply join condition

↓

Combine records into one output

↓

Emit joined stream

↓

End

Join operations combine records from two streams or a stream and a table by matching keys and merging data into a new stream.

Execution Sample

Kafka

KStream<String, String> stream1 = builder.stream("topic1");
KStream<String, String> stream2 = builder.stream("topic2");

KStream<String, String> joined = stream1.join(stream2,
  (v1, v2) -> v1 + ":" + v2,
  JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofSeconds(5)));

joined.to("joined-topic");

This code joins two streams on matching keys within a 5-second window and outputs combined records.

Process Table

Step	Input Stream1 Record	Input Stream2 Record	Join Condition	Output Record
1	key1: A	key1: X	Keys match, within 5s window	key1: A:X
2	key2: B	key3: Y	Keys do not match	No output
3	key1: C	key1: Z	Keys match, within 5s window	key1: C:Z
4	key4: D	key1: W	Keys do not match	No output
5	End of input	End of input	No more records	Stream ends

💡 No more matching keys or records to join, stream processing ends.

Status Tracker

Variable	Start	After 1	After 2	After 3	After 4	Final
stream1 record	none	key1:A	key2:B	key1:C	key4:D	end
stream2 record	none	key1:X	key3:Y	key1:Z	key1:W	end
joined output	none	key1:A:X	none	key1:C:Z	none	end

Key Moments - 2 Insights

Why does no output appear when keys don't match?

Join only emits output when keys from both streams match (see rows 2 and 4 in execution_table). Non-matching keys produce no joined record.

How does the time window affect joining?

Records must arrive within the join window (5 seconds here) to be joined. If outside, no output is produced (implied in join condition column).

Visual Quiz - 3 Questions

Look at the execution_table, what is the output record at step 3?

A key1:A:X

B key1:C:Z

C key2:B:Y

D No output

💡 Hint

Check the Output Record column at step 3 in execution_table.

At which step do the keys not match and no output is produced?

A Step 2

B Step 1

C Step 3

D Step 5

💡 Hint

Look for 'No output' in the Output Record column in execution_table.

If the join window was removed, how would the output change?

A Fewer joined records

B No change in output

C More joined records regardless of time

D Stream would stop immediately

💡 Hint

Join window controls timing; removing it allows all matching keys to join anytime.

Concept Snapshot

Join operations combine two streams or a stream and a table by matching keys.
KStream-KStream join uses a time window to match records.
KStream-KTable join uses latest table state for each key.
Output is a new stream with combined records.
Only matching keys produce output.
Time window controls which records join in streams.

Full Transcript

Join operations in Kafka Streams combine data from two sources by matching keys. For KStream-KStream joins, records must arrive within a time window to be joined. For KStream-KTable joins, the stream record joins with the latest table value for the key. The process starts by reading records from both sources, matching keys, applying join conditions, and producing a combined output record. If keys don't match or records fall outside the time window, no output is produced. The joined results are emitted as a new stream for further processing or output. This visual trace shows step-by-step how records from two streams are joined based on keys and time window, tracking variables and outputs at each step.

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