Why advanced patterns handle complex flows in Kafka - See It in Action

Why advanced patterns handle complex flows

📖 Scenario: You are working in a company that processes orders using Apache Kafka. Orders come from different sources and need to be routed, filtered, and enriched before final processing. Simple Kafka topics and consumers are not enough to handle this complex flow.

🎯 Goal: Build a Kafka Streams application that uses advanced patterns like branching, filtering, and joining streams to handle complex order processing flows.

📋 What You'll Learn

Create a Kafka Streams topology with multiple branches

Filter orders based on status

Join order streams with customer data

Output the processed orders to a final topic

💡 Why This Matters

🌍 Real World

Companies use Kafka Streams to process and enrich real-time data from multiple sources, enabling complex workflows like order processing, fraud detection, and monitoring.

💼 Career

Understanding advanced Kafka patterns is essential for roles like DevOps engineers, data engineers, and backend developers working on scalable event-driven systems.

1

Create initial Kafka Streams topology with input topic

Create a Kafka Streams StreamsBuilder instance called builder and define a stream called ordersStream from the input topic orders.

Kafka

// Create StreamsBuilder and ordersStream from 'orders' topic
// Your code here

Need a hint?

Use new StreamsBuilder() to create the builder. Use builder.stream("orders") to create the stream.

2

Add filtering configuration for order status

Create a String variable called statusFilter and set it to "COMPLETED". This will be used to filter orders by their status.

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream ordersStream = builder.stream("orders");
// Create statusFilter variable
// Your code here

Need a hint?

Declare a String variable named statusFilter and assign it the value "COMPLETED".

3

Filter orders and join with customer data stream

Use ordersStream.filter() with a lambda that keeps only orders where the value contains statusFilter. Then create a KTable called customersTable from the topic customers. Finally, join the filtered orders stream with customersTable on the key using leftJoin.

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream ordersStream = builder.stream("orders");
String statusFilter = "COMPLETED";
// Filter ordersStream by statusFilter and join with customersTable
// Your code here

Need a hint?

Use filter on ordersStream with a lambda checking value.contains(statusFilter). Use builder.table("customers") to create customersTable. Use leftJoin to join filtered orders with customers.

4

Output the enriched orders to the final topic

Send the enrichedOrders stream to the Kafka topic processed-orders using to(). Then build the topology and print its description using builder.build().describe().

Kafka

StreamsBuilder builder = new StreamsBuilder();
KStream ordersStream = builder.stream("orders");
String statusFilter = "COMPLETED";

KStream filteredOrders = ordersStream.filter((key, value) -> value.contains(statusFilter));
KTable customersTable = builder.table("customers");

KStream enrichedOrders = filteredOrders.leftJoin(customersTable, (order, customer) -> order + ", Customer: " + (customer == null ? "Unknown" : customer));
// Send enrichedOrders to 'processed-orders' topic and print topology description
// Your code here

Need a hint?

Use enrichedOrders.to("processed-orders") to send data. Use System.out.println(builder.build().describe()) to print the topology description.