Implementing Stream Joins

1. Understanding Join Types

Join Left Right
Stream-Stream KStream KStream (windowed)
Stream-Table KStream KTable (lookup)
Table-Table KTable KTable

Example: Join overview

stream⋈stream = windowed; stream⋈table = enrichment; table⋈table = materialized

2. Using Stream-Stream Join

Aspect Description Detail
Window Required JoinWindows mandatory Bounded state
Co-Partition Same key + partitions Required
Both Sides Stored Window state stores Buffer

Example: Stream-stream join

clicks.join(impressions,
  (c, i) -> merge(c, i),
  JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)));

3. Using Stream-Table Join

Aspect Description Detail
Lookup Stream record looks up table Enrichment
Trigger Stream side only Table is passive
No Window Point-in-time lookup Latest table value

Example: Enrich orders

orders.join(customers,
  (order, cust) -> order.withName(cust.getName()));

4. Using Table-Table Join

Aspect Description Detail
Materialized Result is a KTable Changelog
Trigger Either side update Both active
Foreign Key Non-key join supported Since 2.4

Example: Table-table join

accounts.join(profiles,
  (acct, profile) -> combine(acct, profile));

5. Implementing Inner Join

Aspect Description Detail
Output Only when both match Intersection
Method join() ValueJoiner
Null No emit if either missing Strict

Example: Inner join

left.join(right, (l, r) -> l + ":" + r);

6. Implementing Left Join

Aspect Description Detail
Output Every left record Right may be null
Method leftJoin() Handle null
Use Optional enrichment Default values

Example: Left join with default

orders.leftJoin(customers,
  (o, c) -> o.withName(c != null ? c.getName() : "UNKNOWN"));

7. Implementing Outer Join

Aspect Description Detail
Output Either side present Union
Supported Stream-stream, table-table Not stream-table
Null Missing side is null Both nullable

Example: Outer join

a.outerJoin(b,
  (x, y) -> (x == null ? "-" : x) + "/" + (y == null ? "-" : y),
  JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)));

8. Configuring Join Windows

Config Description Detail
timeDifference Max time gap Symmetric
before/after Asymmetric bounds One-sided
grace Late record tolerance ofTimeDifferenceAndGrace

Example: Asymmetric window

JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5))
  .before(Duration.ofMinutes(1));

9. Setting Join Retention

Aspect Description Detail
Retention How long window kept ≥ window + grace
State Size Bounded by retention Memory/disk
StreamJoined Configure store retention Per join

Example: Join store config

left.join(right, joiner, windows,
  StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()));

10. Understanding Join Semantics

Aspect Description Detail
Co-Partitioning Same partition count + key Mandatory
Repartition Auto if keys changed selectKey
GlobalKTable No co-partition needed Full replica

Example: GlobalKTable join

stream.join(globalTable,
  (k, v) -> v.getCustomerId(),
  (v, g) -> enrich(v, g));

11. Optimizing Join Performance

Technique Description Benefit
GlobalKTable Avoid repartition Less shuffle
Smaller Windows Less buffered state Memory
Pre-Partition Align keys upstream No repartition

Example: Avoid repartition

// Keep keys aligned to skip internal repartition topics
orders.join(customersTable, (o, c) -> enrich(o, c));

Related Cheat Sheets

Kafka

Monitoring Producer Metrics

Kafka

Configuring Producer Settings

Kafka

Working with Partitions

Kafka

Monitoring Consumer Metrics