Kafka Streams

7 h
requires KFK-07

From your world. dbt joins between a fact and a dimension, but continuous

Enrich orders with customers (KStream-KTable join)

Build a Kafka Streams topology that enriches an orders fact stream with the customers dimension. It is the continuous version of a fact-to-dimension join.

Topics

Topic	Role	Key	Value
`orders`	KStream (facts)	customer id (String)	order amount (String, e.g. `"10.0"`)
`customers`	KTable (dimension)	customer id (String)	customer name (String)
`enriched`	output	customer id (String)	`"

All keys and values use Serdes.String().

Your task — implement `src/main/java/de/commerce/Enricher.java`

Provide a public class de.commerce.Enricher with:

public static Topology build()

It must return a topology that does these steps:

Reads orders as a KStream and customers as a KTable.
Performs a KStream-KTable join. For each order, it looks up the current customer name for that key and emits "<name>|<amount>".
Writes the result to the enriched topic.

Join semantics to follow:

The stream drives output. An order emits exactly one enriched record if the customer key is already in the table.
An order for a customer who is not yet in the table produces no output (an inner KStream-KTable join). If that customer appears later, only future orders are enriched. The join does not emit a past order again.

Grading

The tests use TopologyTestDriver (no broker, no RocksDB, deterministic event time). They send customers and orders in a controlled order. They assert the exact enriched output records. They check that an order before its customer exists yields nothing. They also check that a customer update changes the enrichment of later orders.

This runs with no Docker (vendored JDK + kafka-streams-test-utils).

Starter code

src/main/java/de/commerce/Enricher.java

package de.commerce;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;

public class Enricher {

    public static Topology build() {
        StreamsBuilder builder = new StreamsBuilder();
        // TODO: read `customers` as a KTable and join it to the `orders` stream,
        // emitting "<name>|<amount>" to the `enriched` topic.
        KStream<String, String> orders =
                builder.stream("orders", Consumed.with(Serdes.String(), Serdes.String()));
        // Starter just passes the amount through unchanged (no enrichment) — fails.
        orders.to("enriched", Produced.with(Serdes.String(), Serdes.String()));
        return builder.build();
    }
}

Hints

Hint 1

Build a StreamsBuilder: stream('orders') for the facts, table('customers') for the dimension. Both use Serdes.String().

Hint 2

orders.join(customers, (amount, name) -> name + "|" + amount) does the KStream-KTable lookup; the ValueJoiner receives (streamValue, tableValue).

Hint 3

Write the joined stream to('enriched') and return builder.build(). An order whose customer isn't in the table yet produces no output — that is correct join semantics.

Run it locally

uv run de start KFK-08/01-enrich-orders
# edit workspace/KFK-08/01-enrich-orders/
uv run de test