60-Day Kafka 4 Learning Plan · Week 5 — Day 33 of 60

60-Day Kafka 4 Learning Plan · Week 5 — Kafka Streams Sources: Kafka: The Definitive Guide Ch.11 · kafka.apache.org/documentation/streams/developer-guide/dsl-api#joining

Goal

Learn how to enrich a stream of events with reference data using Kafka Streams joins — KStream-KTable for co-partitioned local state, and KStream-GlobalKTable for fully replicated small reference tables where co-partitioning is impractical — plus the startup-cost and temporal-correctness gotchas both approaches carry in practice.

1. Join types in Kafka Streams

2. KStream-KTable join — enrich with local state

For each order event, look up the user profile from a KTable and merge into an enriched record.

KStream (labs.events, key=userId)
    ↘ join()
KTable (labs.users, key=userId)  ──→  KStream (EnrichedOrderEvent)

// Build user profile KTable from topic
KTable<String, UserProfile> users = builder.table(
    "labs.users",
    Materialized.<String, UserProfile, KeyValueStore<Bytes, byte[]>>
        as("user-profile-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(userProfileSerde)
);

// Inner join: only emits when KTable has an entry for the key
KStream<String, EnrichedOrder> enriched = orders   // key = userId
    .join(
        users,
        (order, profile) -> new EnrichedOrder(order, profile.getTier()),
        Joined.with(Serdes.String(), orderSerde, profileSerde)
    );

// Left join: always emits, profile is null if missing
KStream<String, EnrichedOrder> enrichedLeft = orders
    .leftJoin(
        users,
        (order, profile) -> new EnrichedOrder(order,
            profile != null ? profile.getTier() : "UNKNOWN"),
        Joined.with(Serdes.String(), orderSerde, profileSerde)
    );

enriched.to("labs.enriched-orders");

⚠️ Co-partitioning required: labs.events and labs.users must have the same number of partitions and both must be keyed by userId. Kafka Streams validates this at topology startup and throws TopologyException if violated.

3. inner, left, outer join semantics

Trigger semantics: The join only triggers when the KStream side receives a new event. KTable updates do not trigger new join output — the updated value is used for the next KStream event with that key.

4. GlobalKTable — no co-partitioning needed ★

GlobalKTable replicates all partitions to every Streams instance locally. This means:

Any stream key can look up any table key — no partition alignment needed
The join uses a key extractor lambda to derive the lookup key from the stream value (not necessarily the record key)

// Build product catalog GlobalKTable — replicated fully to every instance
GlobalKTable<String, ProductInfo> products = builder.globalTable(
    "labs.products",
    Materialized.<String, ProductInfo, KeyValueStore<Bytes, byte[]>>
        as("product-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(productInfoSerde)
);

// Join orders (key=userId) with products (key=productId)
// productId != userId — works because GlobalKTable has all partitions locally
KStream<String, EnrichedOrder> enriched = orders     // key = userId
    .join(
        products,
        (orderKey, order) -> order.getProductId(),   // key extractor from VALUE
        (order, product) -> new EnrichedOrder(order, product)
    );

The key extractor (orderKey, order) -> order.getProductId() receives both the record key and value, and returns the key to look up in the GlobalKTable. This is what makes GlobalKTable joins so flexible.

5. KTable vs GlobalKTable — choose wisely

Sizing guidance for GlobalKTable

6. KStream-KStream windowed join (brief)

When joining two event streams (not a table), use a time window to bound which events can match:

KStream<String, PaymentEvent> payments = builder.stream("labs.payments");
KStream<String, OrderEvent> confirmedOrders = builder.stream("labs.confirmed");

// Join orders with payments within 5-minute window
KStream<String, OrderPaymentPair> paired = confirmedOrders.join(
    payments,
    (order, payment) -> new OrderPaymentPair(order, payment),
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
    StreamJoined.with(Serdes.String(), orderSerde, paymentSerde)
);

Both streams must be co-partitioned and keyed by the same join key (e.g. orderId).

7. GlobalKTable startup cost — the trade-off behind “no co-partitioning needed”

GlobalKTable’s flexibility (§4/§5) isn’t free — it comes from every instance holding a full local copy of the entire source topic, which has to be built somehow before the instance is ready to process anything.

Instance startup sequence for a GlobalKTable-using app:
  1. Instance starts
  2. A dedicated GlobalKTable restore thread reads the ENTIRE labs.products
     topic from offset 0, regardless of how large it's grown
  3. Only once fully caught up does the instance begin processing the
     main KStream topology

Practical implications:

Startup time scales with the GlobalKTable’s source topic size — the “< 10 MB, no concern” guidance in §5 is partly about avoiding a slow startup, not just steady-state memory.
Every new instance added during a scale-out event pays this same full-replay cost — it’s not amortized across the fleet the way changelog-topic restoration for a regular (partitioned) KTable state store can be, since a regular KTable instance only restores its assigned slice, while a GlobalKTable instance restores everything regardless of how many instances exist.
This restore thread runs continuously after startup too — it’s not a one-time replay. GlobalKTable data is kept up to date by an always-running background thread separate from the main stream processing, which is why GlobalKTable updates have their own latency characteristics (§8) distinct from a KTable join’s timing.

Why this matters for the sizing table in §5: “1 GB → use KTable or external lookup instead” isn’t just about steady-state RAM — it’s also about how long every single instance takes to become ready to serve traffic on startup or during a rolling deploy, which compounds badly for topics that are actively growing over time.

8. KTable join temporal semantics — a subtle correctness gotcha

A KStream-KTable join always uses whatever the current KTable value is at the moment the KStream event is processed — not the value that was “true” at the KStream event’s own timestamp. This distinction rarely matters for slow-changing reference data (user tier, product catalog) but can silently produce wrong results for fast-changing ones.

t=0: user u1's profile = SILVER  (in the users KTable)
t=1: order event for u1 arrives — joins against SILVER → correct

t=2: user u1 upgrades to GOLD    (KTable updates)
t=3: a DELAYED/late order event for u1, actually placed at t=0.5,
     finally arrives and is processed
     → joins against GOLD (the CURRENT value), not SILVER (what was
       true when the order actually happened)

Why this is easy to miss: it only manifests when the KTable side changes meaningfully and KStream-side events can arrive out of order relative to KTable updates — exactly the kind of edge case that doesn’t show up in a demo with clean, in-order test data, but does show up in production under network delay or a Streams instance restart replaying a backlog. If the join result needs to reflect “what was true when the event happened” rather than “what’s true right now,” a plain KStream-KTable join is the wrong tool — consider a KStream-KStream windowed join (§6) instead, or explicitly version the reference data and carry the relevant version/timestamp in the event itself.

9 Testing joins with TopologyTestDriver

@Test
void innerJoinOnlyEmitsWhenProfileExists() {
    // Populate the KTable side first — pipe a record into the users input topic
    usersInputTopic.pipeInput("u1", userProfile("GOLD"));

    ordersInputTopic.pipeInput("u1", orderEvent());
    ordersInputTopic.pipeInput("u2", orderEvent()); // u2 has no profile — inner join drops it

    var results = enrichedOutputTopic.readKeyValuesToList();
    assertThat(results).hasSize(1);
    assertThat(results.get(0).key).isEqualTo("u1");
}

@Test
void leftJoinEmitsEvenWithoutProfile() {
    ordersInputTopic.pipeInput("u2", orderEvent()); // no profile exists for u2

    var result = enrichedLeftOutputTopic.readValue();
    assertThat(result.tier()).isEqualTo("UNKNOWN"); // per the null-handling logic in §2
}

Ordering matters in the test setup, not just in production. Piping the KTable-side record before the KStream-side event mirrors §3’s trigger semantics — the join only fires on KStream arrival, using whatever KTable state exists at that moment. Reversing the pipe order in a test can produce a different (and misleading) result, which is itself a useful way to build intuition for the §8 temporal gotcha in a controlled setting.

10. Monitoring joins in production

Distinguishing expected vs concerning null-join rates: a small, stable rate of leftJoin misses (new users whose profile hasn’t been created yet) is normal. A rate that’s growing, or that spikes suddenly, points to the upstream reference-data pipeline breaking — apply the same trend-not-absolute-value monitoring philosophy from Day 19 §9 and Day 28 §10 here too.

11. Common pitfalls

Choosing GlobalKTable for a large or fast-growing reference topic “because it’s simpler” — the startup cost (§7) compounds badly at scale, exactly when the simplicity was supposed to help most
Assuming a KStream-KTable join reflects “what was true at event time” — it reflects “what’s true right now,” which silently diverges under out-of-order arrival or replay (§8)
Testing joins without controlling pipe order — the KTable side needs to be populated before the KStream side for the test to reflect realistic trigger semantics (§9)
Treating any leftJoin null rate as acceptable “by design” — a rising trend usually means the reference-data pipeline itself has a problem, not that the join is working as intended (§10)
Not validating co-partitioning before deploying — a topic’s partition count changing independently (Day 12 §12) surfaces here as a startup-time TopologyException, which is disruptive to discover only at deploy time rather than in CI

Key Takeaways

KStream-KTable: enrich events with local state — requires co-partitioning by the same key
GlobalKTable: replicates all partitions everywhere — no co-partitioning, any key field, but every instance pays a full-topic-replay cost on startup and keeps replicating continuously afterward
inner join: only emits when KTable has entry; leftJoin: always emits (null if missing)
GlobalKTable join uses a key extractor lambda — look up by any value field, not just the record key
KTable is partitioned: each instance only holds its assigned slice of the table
Use GlobalKTable for small, slow-growing reference tables (<100 MB); KTable for large co-keyed data
KTable updates do NOT trigger join output — only new KStream events trigger the join, using whatever the KTable’s current value is, not the value that was true at the event’s own timestamp
For correctness needing “value as of event time,” a KStream-KTable join is the wrong tool — consider a windowed KStream-KStream join or explicit versioning instead

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next post.

➡️ Day 34: State stores — RocksDB & interactive queries

Resources

📘 Kafka: The Definitive Guide — Chapter 11 (Joins)
🌐 kafka.apache.org — Joining DSL

Link to Medium blog