KafkaIQ

8.1 Joins in Kafka Streams

Module Overview

Goal: Master joins in Kafka Streams — stream-stream, stream-table, table-table.

What You'll Learn:

  • Join types (inner, left, outer)
  • Stream-Stream joins (windowed)
  • Stream-Table joins
  • Table-Table joins
  • GlobalKTable joins
  • Co-partitioning requirements
  • Join performance considerations

What's NOT Covered Here:

  • KStream/KTable basics (covered in Module 2.1)
  • Windowing details (covered in Module 5.2)
  • State stores (covered in Module 6.1)

Key Insight: Joins combine data from multiple sources — understanding join semantics and windowing is crucial for correctness.

1. Join Overview

Why Joins?

Real-world scenario:

Input 1: Order events (orderId, customerId, amount)
Input 2: Customer data (customerId, name, tier)

Goal: Enrich orders with customer info

Solution: JOIN

Join Types in Kafka Streams

By data type:
├─ Stream-Stream (both event streams)
├─ Stream-Table (stream + changelog)
├─ Table-Table (both changelogs)
└─ Stream-GlobalKTable (stream + replicated table)

By semantics:
├─ Inner (match required)
├─ Left (left side always output)
└─ Outer (both sides output)

2. Co-Partitioning Requirement

What is Co-Partitioning?

Co-partitioning = same key on same partition

Required for:
├─ Stream-Stream joins
├─ Stream-Table joins
└─ Table-Table joins

NOT required for:
└─ Stream-GlobalKTable joins

Co-Partitioning Rules

Topics must have:
1. Same number of partitions
2. Same partitioning strategy
3. Same key type

Example (co-partitioned):
├─ orders: 10 partitions, key = customerId
└─ customers: 10 partitions, key = customerId

Example (NOT co-partitioned):
├─ orders: 10 partitions
└─ customers: 8 partitions  ← Different count!

Auto-Repartitioning

// Kafka Streams auto-repartitions if needed
KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");

// If not co-partitioned, Kafka Streams repartitions automatically
orders.join(customers, ...);

// Repartition topic created: my-app-...-repartition

3. Stream-Stream Joins

Characteristics

Both sides: Event streams (KStream)

Behavior:
├─ Time-bounded (windowed)
├─ Events within time window joined
├─ Stateful (requires state stores)
└─ Produces new event stream (KStream)

Use case:
└─ Correlate related events in time

Example: Click and Impression Events

// Clicks: (adId, clickTime)
KStream<String, Click> clicks = builder.stream("clicks");

// Impressions: (adId, impressionTime)
KStream<String, Impression> impressions = builder.stream("impressions");

// Join clicks with impressions within 10 minutes
KStream<String, ClickImpression> joined = clicks.join(
    impressions,
    (click, impression) -> new ClickImpression(click, impression),  // ValueJoiner
    JoinWindows.of(Duration.ofMinutes(10)),  // Time window
    StreamJoined.with(
        Serdes.String(),  // Key serde
        clickSerde,       // Left value serde
        impressionSerde   // Right value serde
    )
);

joined.to("click-impressions");

How Stream-Stream Joins Work

Time window: 10 minutes

Event 1: Click at 10:00 (adId=123)
└─ Stored in left state store

Event 2: Impression at 10:05 (adId=123)
└─ Looks in left state store for matching clicks [10:00-10:15]
└─ Finds click at 10:00 → emit joined record

Event 3: Click at 10:20 (adId=123)
└─ Looks in right state store for matching impressions [10:10-10:30]
└─ Finds impression at 10:05 (outside window) → no join

Join Semantics: Inner, Left, Outer

// INNER: Only matched events emitted
clicks.join(impressions, joiner, window);
// Output: Only clicks with matching impressions

// LEFT: All left events emitted (right may be null)
clicks.leftJoin(impressions, joiner, window);
// Output: All clicks, impressions optional

// OUTER: All events emitted (either side may be null)
clicks.outerJoin(impressions, joiner, window);
// Output: All clicks and impressions, matched or not

Inner Join Example

Clicks:
├─ 10:00 → click1 (adId=123)
└─ 10:10 → click2 (adId=456)

Impressions:
├─ 10:05 → imp1 (adId=123)
└─ 10:20 → imp2 (adId=789)

Join window: 10 minutes

Output (inner join):
└─ 10:05 → (click1, imp1)  ← Only this matched

click2 and imp2 have no match → not emitted

Left Join Example

Same events as above

Output (left join):
├─ 10:05 → (click1, imp1)   ← Matched
└─ 10:10 → (click2, null)   ← No match, but click emitted

Impressions without clicks not emitted

Outer Join Example

Same events as above

Output (outer join):
├─ 10:00 → (click1, null)   ← Click arrives first
├─ 10:05 → (click1, imp1)   ← Match updates
├─ 10:10 → (click2, null)   ← No match
└─ 10:20 → (null, imp2)     ← Impression no match

All events emitted

Stream-Stream Join State

Join requires TWO state stores:
├─ Left state store (clicks)
└─ Right state store (impressions)

Each record stored for window duration

State store size:
├─ Window size × event rate
├─ Larger window = more state
└─ High event rate = more state

Example:
├─ Window: 10 minutes
├─ Rate: 1000 events/sec
└─ State: ~600,000 records per store

4. Stream-Table Joins

Characteristics

Left side: Event stream (KStream)
Right side: Changelog (KTable)

Behavior:
├─ NOT time-bounded (no window)
├─ Stream event joined with latest table value
├─ Stateful (table backed by state store)
└─ Produces new event stream (KStream)

Use case:
└─ Enrich stream events with reference data

Example: Enrich Orders with Customer Data

// Orders: event stream
KStream<String, Order> orders = builder.stream("orders");

// Customers: changelog (latest customer data)
KTable<String, Customer> customers = builder.table("customers");

// Enrich orders with customer data
KStream<String, EnrichedOrder> enriched = orders.join(
    customers,
    (order, customer) -> new EnrichedOrder(order, customer),
    Joined.with(
        Serdes.String(),
        orderSerde,
        customerSerde
    )
);

enriched.to("enriched-orders");

How Stream-Table Joins Work

Table state (customers):
customerId=1 → Customer("Alice", "Gold")
customerId=2 → Customer("Bob", "Silver")

Stream events (orders):

1. Order(orderId=100, customerId=1, amount=50)
   ├─ Lookup customer=1 in table
   ├─ Found: Customer("Alice", "Gold")
   └─ Emit: EnrichedOrder(100, "Alice", "Gold", 50)

2. Order(orderId=101, customerId=3, amount=75)
   ├─ Lookup customer=3 in table
   ├─ Not found
   └─ No output (inner join)

Join Semantics: Inner vs Left

// INNER: Only orders with matching customers
orders.join(customers, joiner);
// Order without customer → dropped

// LEFT: All orders emitted (customer may be null)
orders.leftJoin(customers, joiner);
// Order without customer → emitted with null customer

Table Updates

Table updates affect FUTURE joins only

Timeline:

10:00 → Customer(1, "Alice", "Gold")
10:01 → Order(100, customerId=1) → Joins with Gold
10:02 → Customer(1, "Alice", "Platinum")  ← Update
10:03 → Order(101, customerId=1) → Joins with Platinum

Previous orders (10:01) NOT retroactively updated

Stream-Table vs Stream-Stream

STREAM-TABLE:
├─ No time window
├─ Joins with LATEST table value
├─ One state store (table only)
└─ Lower state overhead

STREAM-STREAM:
├─ Time window required
├─ Joins with events in window
├─ Two state stores (both sides)
└─ Higher state overhead

Choose based on use case

5. Table-Table Joins

Characteristics

Both sides: Changelogs (KTable)

Behavior:
├─ NOT time-bounded
├─ Join latest values from both tables
├─ Stateful (both tables backed by state stores)
└─ Produces new changelog (KTable)

Use case:
└─ Combine reference data from multiple sources

Example: Join Customers with Addresses

// Customers: (customerId, name, addressId)
KTable<String, Customer> customers = builder.table("customers");

// Addresses: (addressId, street, city)
KTable<String, Address> addresses = builder.table("addresses");

// Change key to addressId for co-partitioning
KTable<String, Customer> customersByAddress = customers
    .toStream()
    .selectKey((k, v) -> v.addressId())
    .toTable();

// Join tables
KTable<String, CustomerWithAddress> joined = customersByAddress.join(
    addresses,
    (customer, address) -> new CustomerWithAddress(customer, address)
);

joined.toStream().to("customers-with-addresses");

How Table-Table Joins Work

Initial state:

Customers table:
customerId=1 → Customer(name="Alice", addressId=100)

Addresses table:
addressId=100 → Address(street="Main St", city="NYC")

Joined table:
addressId=100 → CustomerWithAddress("Alice", "Main St", "NYC")

Update:
Addresses[100] → Address("Broadway", "NYC")

Joined table updated:
addressId=100 → CustomerWithAddress("Alice", "Broadway", "NYC")

Foreign Key Joins

// Customers: (customerId, name, addressId)
// Addresses: (addressId, street, city)

// Foreign key join (customerId → addressId)
KTable<String, CustomerWithAddress> joined = customers.join(
    addresses,
    customer -> customer.addressId(),  // Foreign key extractor
    (customer, address) -> new CustomerWithAddress(customer, address)
);

// Kafka Streams handles repartitioning internally

Table-Table Join Semantics

// INNER: Only records with matches in both tables
customers.join(addresses, joiner);

// LEFT: All left records (right may be null)
customers.leftJoin(addresses, joiner);

// OUTER: All records from both tables
customers.outerJoin(addresses, joiner);

6. GlobalKTable Joins

Why GlobalKTable?

Problem with regular KTable join:
├─ Requires co-partitioning
├─ Must repartition if keys don't match
└─ Overhead for small reference data

Solution: GlobalKTable
├─ ALL partitions consumed by EVERY instance
├─ No co-partitioning required
├─ Join on any field (not just key)

Example: Join with GlobalKTable

// Orders: (orderId, customerId, productId, amount)
KStream<String, Order> orders = builder.stream("orders");

// Products: small reference data (productId, name, price)
GlobalKTable<String, Product> products = builder.globalTable("products");

// Join on productId (not the key)
KStream<String, EnrichedOrder> enriched = orders.join(
    products,
    (orderId, order) -> order.productId(),  // Key mapper (right side)
    (order, product) -> new EnrichedOrder(order, product)
);

enriched.to("enriched-orders");

GlobalKTable vs KTable

KTABLE:
├─ Partitioned (each instance has subset)
├─ Requires co-partitioning for joins
├─ Lower memory (distributed)
└─ Good for large datasets

GLOBAL KTABLE:
├─ Replicated (every instance has all data)
├─ No co-partitioning required
├─ Higher memory (full copy per instance)
└─ Good for small reference data (< 1 GB)

When to Use GlobalKTable

Use GlobalKTable when:
├─ Reference data is small (< 1 GB)
├─ Join on non-key field
├─ Want to avoid repartitioning
└─ Need full dataset on every instance

Avoid GlobalKTable when:
├─ Dataset large (memory constraints)
└─ Data partitioned already

7. Join Performance Considerations

State Store Overhead

Stream-Stream join:
├─ 2 state stores (left + right)
├─ State size = window × event rate
└─ High memory/disk usage

Stream-Table join:
├─ 1 state store (table only)
├─ State size = table size
└─ Lower overhead

Table-Table join:
├─ 2 state stores (both tables)
├─ State size = both table sizes
└─ Medium overhead

Window Size Impact

Stream-Stream join window:

Small window (1 minute):
├─ Less state
├─ Lower memory
└─ Fewer matches

Large window (1 hour):
├─ More state (60× more)
├─ Higher memory
└─ More matches

Choose smallest window that meets requirements

Co-Partitioning Overhead

If not co-partitioned:
├─ Auto-repartition triggered
├─ Repartition topic created
├─ Extra network I/O
├─ Extra disk I/O
└─ Increased latency

Optimize:
└─ Pre-partition input topics correctly

Late-Arriving Data

Stream-Stream joins:

Grace period (Module 5.2):
├─ Allow late records to join
├─ Increases state retention
└─ Trade-off: completeness vs resource usage

After grace period:
└─ Late records dropped or handled separately

8. Join Code Patterns

Pattern 1: Stream-Stream Inner Join

KStream<String, PageView> pageViews = builder.stream("page-views");
KStream<String, AdClick> adClicks = builder.stream("ad-clicks");

KStream<String, ViewClick> joined = pageViews.join(
    adClicks,
    (view, click) -> new ViewClick(view, click),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    StreamJoined.with(Serdes.String(), viewSerde, clickSerde)
);

Pattern 2: Stream-Table Left Join

KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");

KStream<String, EnrichedOrder> enriched = orders.leftJoin(
    customers,
    (order, customer) -> {
        if (customer != null) {
            return new EnrichedOrder(order, customer);
        } else {
            return new EnrichedOrder(order, Customer.unknown());
        }
    }
);

Pattern 3: Table-Table Outer Join

KTable<String, User> users = builder.table("users");
KTable<String, Profile> profiles = builder.table("profiles");

KTable<String, UserProfile> joined = users.outerJoin(
    profiles,
    (user, profile) -> {
        return new UserProfile(
            user != null ? user.name() : "Unknown",
            profile != null ? profile.bio() : "No profile"
        );
    }
);

Pattern 4: GlobalKTable Join

KStream<String, Transaction> transactions = builder.stream("transactions");
GlobalKTable<String, Account> accounts = builder.globalTable("accounts");

KStream<String, EnrichedTransaction> enriched = transactions.join(
    accounts,
    (txnId, txn) -> txn.accountId(),  // Extract right key
    (txn, account) -> new EnrichedTransaction(txn, account)
);

Pattern 5: Multi-Way Join

KStream<String, Order> orders = builder.stream("orders");
KTable<String, Customer> customers = builder.table("customers");
KTable<String, Product> products = builder.table("products");

// First join: orders + customers
KStream<String, OrderCustomer> step1 = orders.join(customers, ...);

// Second join: (orders + customers) + products
// Must change key to productId
KStream<String, FullOrder> step2 = step1
    .selectKey((k, v) -> v.productId())
    .join(products, ...);

9. Join Testing

Unit Testing Joins

TopologyTestDriver testDriver = new TopologyTestDriver(topology, props);

TestInputTopic<String, Order> orderTopic = 
    testDriver.createInputTopic("orders", stringSerde.serializer(), orderSerde.serializer());

TestInputTopic<String, Customer> customerTopic = 
    testDriver.createInputTopic("customers", stringSerde.serializer(), customerSerde.serializer());

TestOutputTopic<String, EnrichedOrder> outputTopic = 
    testDriver.createOutputTopic("output", stringSerde.deserializer(), enrichedSerde.deserializer());

// Send customer first
customerTopic.pipeInput("1", new Customer("Alice"));

// Send order
orderTopic.pipeInput("1", new Order("100", "1", 50));

// Verify join
EnrichedOrder result = outputTopic.readValue();
assertEquals("Alice", result.customerName());

10. Common Join Pitfalls

Pitfall 1: Forgetting Co-Partitioning

// orders: 10 partitions
// customers: 8 partitions

// This will repartition (performance overhead)
orders.join(customers, ...);

// Better: Ensure same partition count upfront

Pitfall 2: Wrong Join Type

// Using inner join when left join needed
orders.join(customers, joiner);
// Orders without customers dropped silently

// Should use left join
orders.leftJoin(customers, joiner);
// All orders emitted

Pitfall 3: Stream-Stream Without Window

// ERROR: Missing window
clicks.join(impressions, joiner);  // Compilation error

// Must specify window
clicks.join(impressions, joiner, JoinWindows.of(Duration.ofMinutes(10)));

Pitfall 4: GlobalKTable for Large Data

// Bad: 10 GB reference data
GlobalKTable<String, LargeData> data = builder.globalTable("large-topic");
// Every instance loads 10 GB → OOM

// Better: Use regular KTable (partitioned)
KTable<String, LargeData> data = builder.table("large-topic");

Pitfall 5: Null Handling

// Left join produces nulls
orders.leftJoin(customers, (order, customer) -> {
    return order.amount() * customer.discount();  // NPE if customer is null!
});

// Correct: Check for null
orders.leftJoin(customers, (order, customer) -> {
    double discount = customer != null ? customer.discount() : 0.0;
    return order.amount() * discount;
});

11. Advanced Join Patterns

Versioned Joins (Kafka 3.5+)

// Use versioned KTable for temporal joins
KTable<String, Customer> customers = builder.table(
    "customers",
    Consumed.with(Serdes.String(), customerSerde),
    Materialized.as("customers-store")
        .withVersionedKeyValueStore()  // Enable versioning
);

// Join with historical customer data at order time
orders.join(customers, ...);
// Joins with customer state at order timestamp

Conditional Joins

// Join only if condition met
KStream<String, Order> largeOrders = orders
    .filter((k, v) -> v.amount() > 1000);

// Join large orders only
largeOrders.join(customers, ...);

Self-Joins

// Join stream with itself (different windows)
KStream<String, Event> events = builder.stream("events");

KStream<String, EventPair> selfJoin = events.join(
    events,
    (event1, event2) -> new EventPair(event1, event2),
    JoinWindows.of(Duration.ofMinutes(5))
);

12. Join Monitoring

Metrics to Monitor

# Join state store size
kafka.streams:type=stream-state-metrics,
              task-id=*,
              store-id=*,
              name=num-entries

# Join rate
kafka.streams:type=stream-processor-node-metrics,
              task-id=*,
              processor-node-id=*,
              name=process-rate

# Late records dropped
kafka.streams:type=stream-processor-node-metrics,
              task-id=*,
              processor-node-id=*,
              name=late-records-drop-rate

13. Mental Model Summary

Join Type Decision Tree

What are you joining?

├─ Two event streams (KStream + KStream)
│  └─ Use Stream-Stream join (windowed)
│
├─ Event stream + reference data (KStream + KTable/GlobalKTable)
│  ├─ Small reference data (< 1 GB)
│  │  └─ Use GlobalKTable (no co-partitioning)
│  └─ Large reference data
│     └─ Use KTable (partitioned)
│
└─ Two reference datasets (KTable + KTable)
   └─ Use Table-Table join

Join Characteristics Summary

STREAM-STREAM:
├─ Time-bounded (window required)
├─ Correlate related events
├─ High state overhead (2 stores)
└─ Example: Click + Impression

STREAM-TABLE:
├─ NOT time-bounded
├─ Enrich stream with latest reference data
├─ Medium state overhead (1 store)
└─ Example: Order + Customer

TABLE-TABLE:
├─ NOT time-bounded
├─ Combine reference datasets
├─ Medium state overhead (2 stores)
└─ Example: User + Profile

STREAM-GLOBALTABLE:
├─ NOT time-bounded
├─ Join on any field (not just key)
├─ High memory (replicated to all instances)
└─ Example: Transaction + small Account lookup

14. Questions You Should Be Able to Answer

After this module, you should be able to answer:

✅ What are the three main join types in Kafka Streams?
✅ When do I need a time window for joins?
✅ What is co-partitioning and why is it required?
✅ When should I use GlobalKTable vs KTable?
✅ What's the difference between inner, left, and outer joins?
✅ How does a Stream-Table join differ from Stream-Stream?
✅ What state stores are created for each join type?
✅ How do table updates affect joins?

15. What's Next?

Now that you understand joins, the next modules will cover:

  • Module 9.1: Exactly-once processing — transactions in joins
  • Module 11.1: Interactive queries — querying join state stores
  • Module 12.1: Monitoring — tracking join performance

Summary

You learned:

  • ✅ Stream-Stream joins are time-bounded (windowed), correlate events
  • ✅ Stream-Table joins enrich streams with latest reference data (not windowed)
  • ✅ Table-Table joins combine reference datasets, output KTable
  • ✅ GlobalKTable joins replicate data to all instances, no co-partitioning needed
  • ✅ Co-partitioning required for all joins except GlobalKTable
  • ✅ Join types: inner (match required), left (left always output), outer (both output)
  • ✅ Window size and grace period affect state size and late data handling
  • ✅ Choose smallest window and lightest join type for performance

Next up: Module 9.1 — Exactly-Once Processing — Learn how Kafka Streams achieves exactly-once semantics including in joins.