Building a Real-Time Event-Driven Order System with Apache Kafka

Over the first 18 parts of this series, we explored the internals of Apache Kafka: from raw disk I/O and Zero-Copy sendfile() transfers up to KRaft consensus and Exactly-Once transactions.

Now it’s time to bring all those concepts together into a complete, production-grade project.

In this capstone tutorial, we will build a Real-Time Event-Driven Order Processing System in Java.

---

Project Architecture & Data Flow

Our system models an e-commerce platform handling incoming orders across five architectural stages:

                                END-TO-END SYSTEM ARCHITECTURE
                                
[ ORDER PRODUCER SERVICE ] 
  | (Idempotent Producer, Avro Schema, acks=all, Key: userId)
  v
[ KAFKA BROKER TOPIC: "order-events" ] (3 Partitions, Replication Factor 3)
  |                                 \
  | (Stream Aggregation)             \ (Consumer Group "fulfillment-group")
  v                                   v
[ KAFKA STREAMS PROCESSOR ]         [ ORDER CONSUMER SERVICE ]
  (5-Min Window Revenue by Category)  (Validates payment & commits offsets)
  |                                   |
  v                                   +---> [ DEAD LETTER TOPIC: "orders-DLT" ] (Malformed records)
[ OUTPUT TOPIC: "category-revenue" ]

---

What We Will Build:

1. Schema Definition: Define binary Avro schemas using Confluent Schema Registry.
2. Resilient Producer Service: Configure an idempotent KafkaProducer with acks = all, custom partitioning, and batch compression.
3. Fault-Tolerant Consumer Service: Implement an At-Least-Once consumer with manual offset commits and Dead Letter Topic (DLT) error routing.
4. Real-Time Stream Processor: Build a Kafka Streams pipeline computing 5-minute sliding window revenue totals per product category backed by embedded RocksDB state.

---

Step 1: Define the Avro Data Contract (OrderEvent.avsc)

Create the Avro schema file in src/main/avro/OrderEvent.avsc:

{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.company.kafka.events",
  "doc": "Schema for e-commerce order creation events",
  "fields": [
    { "name": "orderId", "type": "string" },
    { "name": "userId", "type": "string" },
    { "name": "category", "type": "string" },
    { "name": "itemCount", "type": "int" },
    { "name": "totalAmount", "type": "double" },
    { "name": "timestamp", "type": "long" }
  ]
}

---

Step 2: Build the Resilient Producer (OrderProducerService.java)

Our producer enforces strict durability rules (acks = all, min.insync.replicas = 2), enables idempotence to eliminate duplicate network retries, and keys messages by userId to preserve per-user event ordering.

package com.company.kafka.producer;

import com.company.kafka.events.OrderEvent;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class OrderProducerService {
    private static final Logger logger = LoggerFactory.getLogger(OrderProducerService.class);
    private final Producer<String, OrderEvent> producer;
    private final String topicName;

    public OrderProducerService(String bootstrapServers, String schemaRegistryUrl, String topicName) {
        this.topicName = topicName;
        
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put("schema.registry.url", schemaRegistryUrl);

        // Durability & Reliability Configurations
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true"); // Prevents out-of-order retries!
        props.put(ProducerConfig.ACKS_CONFIG, "all");                // Requires full ISR acknowledgment!
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); // Retry until delivery timeout
        
        // High-Throughput Batching Configurations
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);          // 64KB Batch size
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);              // Wait up to 20ms to fill batches
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");   // ZSTD batch compression

        this.producer = new KafkaProducer<>(props);
    }

    public void sendOrder(OrderEvent event) {
        // Use userId as partition key to guarantee ordering for that user!
        ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(topicName, event.getUserId().toString(), event);

        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                logger.info("Sent OrderId={} to Partition={} at Offset={}", 
                        event.getOrderId(), metadata.partition(), metadata.offset());
            } else {
                logger.error("Failed to deliver OrderId={}", event.getOrderId(), exception);
            }
        });
    }

    public void close() {
        producer.close();
    }
}

---

Step 3: Real-Time Stream Analytics (RevenueStreamsProcessor.java)

Next, we build a real-time stream processing pipeline using Kafka Streams.

This service reads the order-events stream, groups events by product category, and computes total revenue over a 5-minute sliding window.

package com.company.kafka.streams;

import com.company.kafka.events.OrderEvent;
import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

import java.time.Duration;
import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class RevenueStreamsProcessor {
    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092";
        String schemaRegistryUrl = "http://localhost:8081";

        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "revenue-analytics-processor");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class.getName());
        props.put("schema.registry.url", schemaRegistryUrl);

        // Configure Specific Avro Serde for OrderEvent
        Map<String, String> serdeConfig = Collections.singletonMap("schema.registry.url", schemaRegistryUrl);
        SpecificAvroSerde<OrderEvent> orderEventSerde = new SpecificAvroSerde<>();
        orderEventSerde.configure(serdeConfig, false);

        StreamsBuilder builder = new StreamsBuilder();

        // 1. Read input KStream from "order-events" topic
        KStream<String, OrderEvent> ordersStream = builder.stream("order-events", Consumed.with(Serdes.String(), orderEventSerde));

        // 2. Group by product category -> Window into 5-minute tumbling windows -> Sum total revenue
        KTable<Windowed<String>, Double> categoryRevenue = ordersStream
                .groupBy((userId, event) -> event.getCategory().toString(), Grouped.with(Serdes.String(), orderEventSerde))
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
                .aggregate(
                        () -> 0.0, // Initial accumulator value
                        (category, event, currentTotal) -> currentTotal + event.getTotalAmount(),
                        Materialized.with(Serdes.String(), Serdes.Double()) // Embedded RocksDB state store!
                );

        // 3. Write aggregated revenue metrics to output topic
        categoryRevenue.toStream()
                .map((windowedKey, total) -> new KeyValue<>(
                        windowedKey.key() + "@" + windowedKey.window().startTime().getEpochSecond(), 
                        "Total Revenue: $" + total))
                .to("category-revenue", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

---

Step 4: Build the Fault-Tolerant Consumer with DLT Routing (OrderConsumerService.java)

Our consumer service handles order fulfillment. It disables automatic commits, processes records with At-Least-Once guarantees, and routes corrupted records to a Dead Letter Topic (orders-DLT).

package com.company.kafka.consumer;

import com.company.kafka.events.OrderEvent;
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.*;

public class OrderConsumerService implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(OrderConsumerService.class);
    private final KafkaConsumer<String, OrderEvent> consumer;
    private final KafkaProducer<String, String> dltProducer;
    private final String dltTopic = "orders-DLT";
    private volatile boolean running = true;

    public OrderConsumerService(String bootstrapServers, String schemaRegistryUrl, String groupId, String topicName) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class.getName());
        props.put("schema.registry.url", schemaRegistryUrl);

        // At-Least-Once Manual Commit Configurations
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false"); // Manual commits!
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        
        // Cooperative Sticky Rebalancing
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
                CooperativeStickyAssignor.class.getName());

        this.consumer = new KafkaConsumer<>(props);
        this.consumer.subscribe(Collections.singletonList(topicName));

        // DLT Producer setup
        Properties dltProps = new Properties();
        dltProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        dltProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        dltProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        this.dltProducer = new KafkaProducer<>(dltProps);
    }

    @Override
    public void run() {
        try {
            while (running) {
                ConsumerRecords<String, OrderEvent> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, OrderEvent> record : records) {
                    try {
                        processFulfillment(record.value());
                    } catch (Exception e) {
                        logger.error("Error processing record at Offset={}. Routing to DLT!", record.offset(), e);
                        sendToDeadLetterTopic(record, e.getMessage());
                    }
                }
                
                // Commit offsets asynchronously AFTER processing batch
                if (!records.isEmpty()) {
                    consumer.commitAsync((offsets, exception) -> {
                        if (exception != null) {
                            logger.error("Failed to commit offsets asynchronously: {}", offsets, exception);
                        }
                    });
                }
            }
        } finally {
            try {
                consumer.commitSync(); // Final sync commit on shutdown!
            } finally {
                consumer.close();
                dltProducer.close();
            }
        }
    }

    private void processFulfillment(OrderEvent event) {
        if (event.getTotalAmount() <= 0) {
            throw new IllegalArgumentException("Invalid order amount: " + event.getTotalAmount());
        }
        logger.info("Successfully fulfilled OrderId={} for Amount=${}", event.getOrderId(), event.getTotalAmount());
    }

    private void sendToDeadLetterTopic(ConsumerRecord<String, OrderEvent> record, String errorMessage) {
        String payload = String.format("{\"error\":\"%s\", \"failedOffset\":%d}", errorMessage, record.offset());
        dltProducer.send(new ProducerRecord<>(dltTopic, record.key(), payload));
    }

    public void stop() {
        this.running = false;
    }
}

---

Step 5: Verification & Production Runbook

To verify our end-to-end system in a local environment:

1. Start Infrastructure via Docker Compose

version: '3.8'
services:
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:29092,CONTROLLER://0.0.0.0:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'

  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAPSERVERS: 'kafka:29092'

2. Inspect Consumer Lag & DLT Topics

Run Kafka CLI commands to verify consumer group health and inspect the Dead Letter Topic:

# Check Consumer Group Lag
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --describe --group fulfillment-group

# Read Dead Letter Topic messages
kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic orders-DLT --from-beginning

---

Summary of Applied Concepts

In this single project, we combined:

• Part 04: Sharding topics by entity keys (userId).
• Part 07: High-throughput producer batching (linger.ms = 20, zstd compression).
• Part 08–09: Idempotent delivery (enable.idempotence = true) and quorum durability (acks = all).
• Part 10–11: Pull-based consumer scaling, manual offset commits, and At-Least-Once delivery.
• Part 12: Cooperative Sticky Rebalancing (CooperativeStickyAssignor).
• Part 16: Schema enforcement via Avro and Schema Registry.
• Part 17: Real-time stream processing with Kafka Streams and embedded RocksDB state.

References & Further Reading

1. Kreps, J., Narkhede, N., & Rao, J. (2011). Kafka: a Distributed Messaging System for Log Processing. NetDB Workshop.
2. Apache Kafka Wiki. Apache Kafka KIP-98 & KIP-500 Specifications. Kafka Improvement Proposals.
3. Narkhede, N., et al. (2021). Kafka: The Definitive Guide (2nd Edition). O’Reilly Media.

---