Real-Time Order Pipeline with Kafka

Build a local Kafka pipeline with a producer and three consumer microservices.

Introduction

30 Second Summary

When you order something online, multiple systems react to that single event at once. Payment gets charged, fraud checks run, and inventory updates. These services all need to see the same data independently and in real time.

In this project, you will build a local real-time event streaming pipeline using Apache Kafka and Python. A producer publishes simulated order events, and three independent consumer microservices each process the same stream differently.

What You'll Build

A local event streaming pipeline where a Python producer publishes order events to a Kafka topic, and three consumer microservices independently process the same stream for payments, fraud detection, and analytics.

By the end of this project, you'll have:

  • A running Kafka broker in Docker that you can interact with from the command line.
  • A Python producer script generating simulated e-commerce order events with key-based partitioning.
  • Three independent consumer microservices (payments, fraud detection, analytics) demonstrating the fan-out pattern.
  • Secret Mission: A dead-letter topic for failed fraud checks and a burst producer that stress-tests your pipeline with high-volume traffic.

Are there any prerequisites?

No prior Kafka experience is needed. You should be comfortable running commands in a terminal and reading basic Python code. This guide includes setup for all required tools.

Before We Start

Before We Start...

Set Up Python and Start Kafka

Every event streaming pipeline needs two things: a broker to store and route events, and a client to produce and consume them. In this project, Apache Kafka is your broker and Python is your client.

In this step, you'll install Python, set up a clean project environment, and start a local Kafka broker using Docker. By the end, everything will be running and ready for code.

In this step, get ready to:
  • Install Python 3.14 and create a project directory with a virtual environment.
  • Install the confluent-kafka Python library.
  • Start a Kafka broker in Docker.
Install Python and set up your project

This project uses Python 3.14 to interact with Kafka. Let's check whether it's already installed on your machine.

  • Open your terminal by pressing Cmd+Space to open Spotlight search.
  • Type Terminal and press Enter to open it.
  • Check your Python version by running this command:
python3 --version

✔️ I see version 3.14

Python 3.14 is ready. You can move on to creating your project directory below.

ⓧ I see an older version

Your Mac has an older Python installed (likely from the system or a previous install). You need Python 3.14 for this project.

  • Download the macOS installer from the Python release page:
  • https://www.python.org/downloads/release/python-3146/
  • Click Download macOS installer and run the downloaded file.
  • Follow the installer prompts and accept the defaults.
  • Once complete, close and reopen Terminal, then run the version check again:
python3 --version

You should now see Python 3.14.6 or similar.

Still seeing the old version?

The python.org installer places symlinks in /usr/local/bin/. If your shell still resolves to the older system Python, try using the versioned command instead:

python3.14 --version

If that works, use python3.14 in place of python3 for the rest of this project.

ⓧ Command not found

Python is not installed on your machine. Let's install it from python.org.

  • Download the macOS installer from the Python release page:
  • https://www.python.org/downloads/release/python-3146/
  • Click Download macOS installer and run the downloaded file.
  • Follow the installer prompts and accept the defaults.
  • Once complete, close and reopen Terminal, then confirm the install:
python3 --version

You should see Python 3.14.6 or similar.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir kafka-pipeline
cd kafka-pipeline
  • Confirm you're in the right place by running:
pwd

You should see /Users/<your-username>/Desktop/kafka-pipeline in the output.

A virtual environment isolates this project's Python packages from the rest of your system. This prevents version conflicts between projects.

  • Create and activate a virtual environment by running this command:
python3 -m venv .venv && source .venv/bin/activate

You should see (.venv) appear at the beginning of your terminal prompt. This confirms you're now working inside the virtual environment.

Don't see (.venv) in your prompt?

Make sure you used python3 (not python) in the command. If you installed Python 3.14 fresh and your shell still points to an older version, try:

python3.14 -m venv .venv && source .venv/bin/activate

If you see a permission error, make sure you're inside the kafka-pipeline folder you just created. Help me troubleshoot my virtual environment setup.

Install the confluent-kafka library

The confluent-kafka library is a high-performance Python client for Apache Kafka. It's built on top of a C library called librdkafka, which makes it much faster than pure-Python alternatives.

  • Install the library by running this command:
pip install confluent-kafka==2.15.0

Why pin the version?

Pinning to confluent-kafka==2.15.0 ensures everyone following this project gets the same behaviour. Library updates can change APIs or introduce bugs, so locking the version keeps your project reproducible.

  • Verify the library is installed correctly by running:
python3 -c "import confluent_kafka; print(confluent_kafka.version())"

You should see a version tuple printed in the terminal (e.g. a tuple starting with 2.15.0). This confirms the library is importable and ready to use.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is active. You should see (.venv) in your terminal prompt. If not, activate it again:

source .venv/bin/activate

Then re-run the pip install command. Help me fix my confluent-kafka installation.

Start a Kafka broker in Docker

Apache Kafka is a Java application that traditionally needed a separate coordination service called ZooKeeper. Modern Kafka (4.x) runs in KRaft mode, which bundles everything into a single process. Docker lets you start this entire broker with one command.

  • Make sure Docker Desktop is running. Press Cmd+Space, type Docker, and press Enter to open it.
  • Wait until you see the Docker Desktop dashboard (this may take a few seconds while the engine starts).
  • Return to your terminal and start a Kafka broker by running this command:
docker run -d --name kafka -p 9092:9092 apache/kafka:4.3.1

What does this command do?

  • docker run creates and starts a new container.
  • -d runs the container in detached mode (in the background).
  • --name kafka gives the container the name kafka so you can reference it easily.
  • -p 9092:9092 maps port 9092 inside the container to port 9092 on your machine, letting your Python scripts connect.
  • apache/kafka:4.3.1 is the official Kafka Docker image running in KRaft mode (no ZooKeeper needed).

The first time you run this, Docker will download the Kafka image. This can take a minute or two depending on your connection.

  • Once the command completes, verify Kafka is running by checking its logs:
docker logs kafka

Look for Kafka Server started in the output. You'll see many INFO lines scrolling past. The important one is near the end.

Give it a moment

If you don't see the started message immediately, wait 5-10 seconds and run docker logs kafka again. Kafka needs a few seconds to initialize.

Seeing an error from Docker?

If you see a "name already in use" error, a container named kafka already exists from a previous attempt. Remove it first:

docker rm -f kafka

Then re-run the docker run command above. If you see a "Cannot connect to the Docker daemon" error, Docker Desktop is not running yet. Open it and wait for the engine to start before trying again. Help me troubleshoot my Kafka Docker container.

Your environment is ready. You have Python, a virtual environment with the Kafka client library, and a running broker. Next up, you'll create a Kafka topic and explore how the broker organizes data into partitions.

Create Topics and Explore the Broker

Your Kafka broker is running and ready to accept connections. Before writing any Python code, you need a place for your order events to live.

In Kafka, events are organized into topics. A topic is a named stream of events that producers write to and consumers read from. In this step, you'll create the orders topic and confirm that you can send and receive messages through it.

In this step, get ready to:
  • Create the orders topic with 3 partitions.
  • Send and read a test message using Kafka's built-in CLI tools.
  • Understand topics, partitions, and event durability.
Create the orders topic with 3 partitions

A Kafka topic needs to exist before producers can write to it. You'll also configure the topic with multiple partitions, which allows multiple consumers to read from the same topic in parallel later on.

  • Create the orders topic by running this command:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic orders --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092

What does this command do?

  • docker exec kafka runs a command inside your running Kafka container.
  • /opt/kafka/bin/kafka-topics.sh is the Kafka CLI tool for managing topics.
  • --create --topic orders creates a new topic named orders.
  • --partitions 3 splits the topic into 3 partitions so multiple consumers can read in parallel.
  • --replication-factor 1 keeps one copy of the data (since you only have one broker).
  • Verify the topic was created and inspect its partition layout by running:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092

You should see output showing PartitionCount:3 and ReplicationFactor:1, followed by three lines describing each partition's leader and replicas.

Seeing "Topic 'orders' already exists"?

This means the topic was already created from a previous attempt. You can safely proceed to the describe command to confirm its configuration.

If you need to start fresh, delete the topic first and then re-run the create command:

docker exec kafka /opt/kafka/bin/kafka-topics.sh --delete --topic orders --bootstrap-server localhost:9092

Help me troubleshoot this Kafka topic creation error.

Send and read a test message with the console tools

Kafka ships with built-in console tools that let you produce and consume messages from the command line. You'll use these to confirm end-to-end connectivity before writing Python code.

  • Start the console producer by running this command:
docker exec -it kafka /opt/kafka/bin/kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092

What happens next?

The console producer opens an interactive prompt (you'll see a > character). Each line you type becomes a separate event published to the orders topic.

  • Type the following test message and press Enter:
test order message
  • Press Ctrl+C to exit the console producer.
  • Now read the message back by starting the console consumer:
docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh --topic orders --from-beginning --bootstrap-server localhost:9092

You should see test order message printed in your terminal. This confirms your broker can receive and deliver events end-to-end.

  • Press Ctrl+C to exit the console consumer.

Not seeing your test message?

Make sure you included --from-beginning in the consumer command. Without it, the consumer only reads new messages arriving after it starts.

If you see no output at all, confirm your Kafka container is still running with docker ps. Look for the container named kafka in the list.

Help me figure out why my test message isn't showing up.

Understand what you just built

What is a topic, and why partitions?

A topic is a named stream of events. Think of it like a log file that multiple services can read from independently. The orders topic you created is where all order events will flow through.

Partitions split a topic into parallel lanes. With 3 partitions, up to 3 consumers in the same group can each read from a different partition simultaneously. This is how Kafka scales to handle high throughput.

The broker stores events durably on disk. Even after your console consumer read the test message, the event still exists in the topic. New consumers can read it again from the beginning at any time.

Why 3 partitions for this project?

You'll build 3 consumer services in Step 4. Each consumer uses a different consumer group, so each independently reads all 3 partitions. Later in Step 5, you'll see what happens when two consumers share the same group. Kafka rebalances the 3 partitions between them automatically.

Your topic is ready and you've confirmed messages flow end-to-end. Next up, you'll write a Python producer that generates realistic order events and publishes them to this topic.

Build the Order Producer

Your Kafka broker is live and the orders topic is waiting for data. Now it's time to write the code that fills it.

In this step, you'll build a Python producer that generates simulated e-commerce order events and publishes them to Kafka. Each event represents a customer placing an order with details like product, amount, and city.

In this step, get ready to:
  • Connect a Python producer to your Kafka broker.
  • Generate and publish random order events to the orders topic.
  • Verify events are flowing by reading them with a console consumer.
Configure the producer and define the delivery callback

A Kafka producer needs two things: a connection to the broker, and a way to confirm messages were delivered. You'll configure both, along with sample data for generating random orders.

  • Create a file called producer.py in your project folder by running:
touch producer.py
  • Confirm the file exists by running:
ls

You should see producer.py in the output.

  • Open producer.py in your text editor, then paste the following code:
import json
import random
import time
from confluent_kafka import Producer

# Connection settings for the Kafka broker
conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'order-producer'
}

# Create the producer instance
producer = Producer(conf)

# Sample data for generating random orders
products = ['laptop', 'mouse', 'keyboard', 'monitor', 'headset']
cities = ['Chicago', 'New York', 'Dallas', 'San Jose', 'Seattle']


# Callback that fires once the broker acknowledges each message
def delivery_report(err, msg):
    if err is not None:
        print(f'Delivery failed: {err}')
    else:
        print(f'Order delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}')

What does this code do?

The conf dictionary tells the producer where to find your Kafka broker (localhost:9092) and assigns a client ID for identification in broker logs.

The products and cities lists are pools of sample data. Each order will randomly pick from these.

The delivery_report function is a callback. Kafka calls it after each message is either successfully written or fails. This gives you real-time visibility into whether your events are reaching the broker.

  • Save producer.py.

Seeing an import error?

Make sure your virtual environment is active. You should see (.venv) at the start of your terminal prompt. If not, reactivate it by running source .venv/bin/activate.

Help me troubleshoot this import error

Build the order generation loop

The producer needs to continuously generate order events and send them to Kafka. You'll use a loop that creates a random order every second and publishes it with the city field as the message key.

  • Add the following code to the bottom of producer.py, below the delivery_report function:
print('Order Producer started. Press Ctrl+C to stop.')

try:
    while True:
        # Build a random order event
        order = {
            'order_id': random.randint(1000, 9999),
            'product': random.choice(products),
            'amount': round(random.uniform(50, 2000), 2),
            'city': random.choice(cities),
            'timestamp': time.time()
        }

        # Publish the order to Kafka, using city as the message key
        producer.produce(
            topic='orders',
            key=order['city'],
            value=json.dumps(order),
            callback=delivery_report
        )

        # Trigger delivery report callbacks without blocking
        producer.poll(0)
        time.sleep(1)

except KeyboardInterrupt:
    print('\nShutting down producer...')
finally:
    # Ensure all buffered messages are sent before exiting
    producer.flush()
    print('Producer shut down.')

What does this code do?

The while True loop generates one random order per second. Each order is a dictionary with five fields: order_id, product, amount, city, and timestamp.

The producer.produce() call sends the order to Kafka. The key=order['city'] argument ensures all orders from the same city always land in the same partition. This is key-based partitioning.

The producer.poll(0) call triggers pending delivery report callbacks without blocking. The producer.flush() in the finally block ensures all buffered messages are sent before the script exits.

Why use city as the message key?

Kafka uses the message key to decide which partition receives each message. All messages with the same key always go to the same partition. This guarantees ordering within a city.

In production, this means you could process all New York orders on the same consumer thread, maintaining the exact sequence those orders arrived in.

  • Save producer.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

import json
import random
import time
from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'client.id': 'order-producer'
}

producer = Producer(conf)

products = ['laptop', 'mouse', 'keyboard', 'monitor', 'headset']
cities = ['Chicago', 'New York', 'Dallas', 'San Jose', 'Seattle']


def delivery_report(err, msg):
    if err is not None:
        print(f'Delivery failed: {err}')
    else:
        print(f'Order delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}')


print('Order Producer started. Press Ctrl+C to stop.')

try:
    while True:
        order = {
            'order_id': random.randint(1000, 9999),
            'product': random.choice(products),
            'amount': round(random.uniform(50, 2000), 2),
            'city': random.choice(cities),
            'timestamp': time.time()
        }

        producer.produce(
            topic='orders',
            key=order['city'],
            value=json.dumps(order),
            callback=delivery_report
        )

        producer.poll(0)
        time.sleep(1)

except KeyboardInterrupt:
    print('\nShutting down producer...')
finally:
    producer.flush()
    print('Producer shut down.')
Run the producer and verify events are flowing

Time to see your producer in action. You'll start it in one terminal and read its output from a second terminal using Kafka's built-in console consumer.

  • Run the producer by executing:
python3 producer.py

You should see delivery confirmations printing every second, showing the topic, partition number, and offset for each order.

What should I see?

Your terminal should display lines like:

Order Producer started. Press Ctrl+C to stop.
Order delivered to orders [1] at offset 0
Order delivered to orders [0] at offset 0
Order delivered to orders [2] at offset 0

The partition numbers ([0], [1], [2]) vary based on which city key the message carries. Offsets increment as more messages land in each partition.

  • Open a second terminal window (keep the producer running in the first one).
  • Navigate to your project folder in the new terminal.
  • Activate your virtual environment in the new terminal by running:
source .venv/bin/activate
  • Read the events from the orders topic by running:
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh --topic orders --from-beginning --bootstrap-server localhost:9092

You should see JSON order events streaming in your second terminal. Each line is one order event that your producer published.

What should I see?

The console consumer prints the raw message values. You'll see JSON objects like:

{"order_id": 4821, "product": "laptop", "amount": 1234.56, "city": "Chicago", "timestamp": 1751347200.0}

Each line is a separate order event. The data is random, so your exact values will differ.

  • Stop the console consumer with Ctrl+C in the second terminal.
  • Leave the producer running in the first terminal for now.

Not seeing any messages?

Make sure the producer is still running in your first terminal. Check for delivery confirmations there. If you see Delivery failed messages, the Kafka broker might have stopped.

Verify the broker is running with docker ps. If the kafka container is not listed, restart it with docker start kafka.

Help me debug this connection issue

Your producer is live and streaming order events into Kafka. Next up, you'll build three consumer services that each process these same events in a completely different way.

Build Three Consumer Microservices

Your producer is publishing order events into the orders topic. Now you need services that actually do something with those events.

In a real e-commerce system, multiple teams need the same order data for different purposes. The payments team charges the customer, the fraud team checks for suspicious activity, and the analytics team tracks sales. Each team runs its own service that reads from the same stream independently.

In this step, get ready to:
  • Create a payments consumer that logs charges for each order.
  • Create a fraud detection consumer that flags suspicious orders.
  • Create an analytics consumer that tracks live sales counts and revenue.
Create the payments consumer

Each consumer service connects to the same orders topic but uses a different group.id. This is what gives each service its own independent copy of every message.

  • Create a new file called consumer_payments.py in your project folder and add the following code:
import json
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'payments-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

print('Payments Service started. Press Ctrl+C to stop.')

try:
    while True:
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        order = json.loads(msg.value().decode('utf-8'))
        print(f'[PAYMENTS] Charging ${order["amount"]:.2f} for {order["product"]} '
              f'(Order #{order["order_id"]} from {order["city"]})')

except KeyboardInterrupt:
    print('\nShutting down payments service...')
finally:
    consumer.close()
    print('Payments service shut down.')

What does this code do?

This consumer connects to your Kafka broker and subscribes to the orders topic.

  • group.id set to payments-group gives this consumer its own independent position in the stream. No other consumer with a different group ID will interfere with it.
  • auto.offset.reset set to earliest means the consumer starts reading from the very beginning of the topic when it first joins.
  • consumer.poll(1.0) waits up to 1 second for a new message. If nothing arrives, it returns None and the loop continues.
  • consumer.close() cleanly leaves the consumer group when you press Ctrl+C, so Kafka knows this consumer is gone.

What is a consumer group?

A consumer group is a set of consumers that share the same group.id. Kafka treats them as one logical reader. If two consumers share a group ID, Kafka splits the partitions between them so each message is only processed once.

When consumers have different group IDs (like your three services will), each group gets its own complete copy of every message. This is the fan-out pattern.

  • Save consumer_payments.py.
  • Test the consumer by running:
python3 consumer_payments.py
  • You should see Payments Service started. Press Ctrl+C to stop. printed in your terminal. If you sent test messages in earlier steps, you'll also see [PAYMENTS] lines appear for each order.
  • Press Ctrl+C to stop the consumer for now.

Seeing a connection error?

  • Make sure your Kafka container is still running. Check with docker ps and look for the kafka container.
  • Verify your virtual environment is active. You should see (.venv) at the start of your terminal prompt.
  • Check that bootstrap.servers is set to localhost:9092 in your config dictionary.

Help me debug my Kafka consumer connection error

✔️ Awesome, I've got everything!

Great. Make sure you've saved consumer_payments.py and stopped the consumer with Ctrl+C before moving on.

ⓧ I'd like to double check the full code

import json
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'payments-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

print('Payments Service started. Press Ctrl+C to stop.')

try:
    while True:
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        order = json.loads(msg.value().decode('utf-8'))
        print(f'[PAYMENTS] Charging ${order["amount"]:.2f} for {order["product"]} '
              f'(Order #{order["order_id"]} from {order["city"]})')

except KeyboardInterrupt:
    print('\nShutting down payments service...')
finally:
    consumer.close()
    print('Payments service shut down.')
Create the fraud detection consumer

The fraud detection service reads the same stream of orders but applies business rules to flag suspicious activity. It checks two conditions: orders above $1500 and orders from cities on a high-risk list.

  • Create a new file called consumer_fraud.py in your project folder and add the following code:
import json
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'fraud-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

HIGH_RISK_CITIES = {'Seattle', 'Dallas'}
AMOUNT_THRESHOLD = 1500

print('Fraud Detection Service started. Press Ctrl+C to stop.')

try:
    while True:
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        order = json.loads(msg.value().decode('utf-8'))

        is_high_amount = order['amount'] > AMOUNT_THRESHOLD
        is_high_risk_city = order['city'] in HIGH_RISK_CITIES

        if is_high_amount or is_high_risk_city:
            reasons = []
            if is_high_amount:
                reasons.append(f'amount ${order["amount"]:.2f} exceeds ${AMOUNT_THRESHOLD}')
            if is_high_risk_city:
                reasons.append(f'high-risk city: {order["city"]}')
            print(f'[FRAUD ALERT] Order #{order["order_id"]}: {", ".join(reasons)}')
        else:
            print(f'[FRAUD OK] Order #{order["order_id"]}: ${order["amount"]:.2f} from {order["city"]}')

except KeyboardInterrupt:
    print('\nShutting down fraud detection service...')
finally:
    consumer.close()
    print('Fraud detection service shut down.')

What does this code do?

This consumer uses the same polling pattern as the payments service, but adds fraud detection logic after deserializing each order.

  • HIGH_RISK_CITIES is a Python set containing Seattle and Dallas. Sets provide fast membership checks.
  • AMOUNT_THRESHOLD is set to 1500. Any order above this amount triggers a fraud alert.
  • The service builds a list of reasons for each flagged order. An order can trigger both conditions (high amount AND high-risk city), and the output shows all applicable reasons.
  • group.id is fraud-group. This is different from payments-group, so this consumer gets its own independent copy of every message.
  • Save consumer_fraud.py.
  • Test the fraud consumer by running:
python3 consumer_fraud.py
  • You should see Fraud Detection Service started. Press Ctrl+C to stop. printed in your terminal. If there are existing orders in the topic, you'll see a mix of [FRAUD ALERT] and [FRAUD OK] lines.
  • Press Ctrl+C to stop the consumer.

Not seeing any FRAUD ALERT lines?

  • This is normal if the random orders generated earlier happened to be below $1500 and not from Seattle or Dallas. You'll see fraud alerts once you run the full pipeline with more data in the next step.
  • If you see no output at all (not even the startup message), check for syntax errors in your file. Make sure the f-string quotes match correctly.

Help me debug my fraud detection consumer

✔️ Awesome, I've got everything!

Great. Make sure you've saved consumer_fraud.py and stopped the consumer before moving on.

ⓧ I'd like to double check the full code

import json
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'fraud-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

HIGH_RISK_CITIES = {'Seattle', 'Dallas'}
AMOUNT_THRESHOLD = 1500

print('Fraud Detection Service started. Press Ctrl+C to stop.')

try:
    while True:
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        order = json.loads(msg.value().decode('utf-8'))

        is_high_amount = order['amount'] > AMOUNT_THRESHOLD
        is_high_risk_city = order['city'] in HIGH_RISK_CITIES

        if is_high_amount or is_high_risk_city:
            reasons = []
            if is_high_amount:
                reasons.append(f'amount ${order["amount"]:.2f} exceeds ${AMOUNT_THRESHOLD}')
            if is_high_risk_city:
                reasons.append(f'high-risk city: {order["city"]}')
            print(f'[FRAUD ALERT] Order #{order["order_id"]}: {", ".join(reasons)}')
        else:
            print(f'[FRAUD OK] Order #{order["order_id"]}: ${order["amount"]:.2f} from {order["city"]}')

except KeyboardInterrupt:
    print('\nShutting down fraud detection service...')
finally:
    consumer.close()
    print('Fraud detection service shut down.')
Create the analytics consumer

The analytics service tracks running totals across all orders. It counts how many of each product have been sold and accumulates total revenue. Every time a new order arrives, it prints updated live totals.

  • Create a new file called consumer_analytics.py in your project folder and add the following code:
import json
from collections import defaultdict
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'analytics-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

sales_count = defaultdict(int)
total_revenue = 0.0

print('Analytics Service started. Press Ctrl+C to stop.')

try:
    while True:
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        order = json.loads(msg.value().decode('utf-8'))

        sales_count[order['product']] += 1
        total_revenue += order['amount']

        print(f'[ANALYTICS] Live Sales Count: {dict(sales_count)} | '
              f'Total Revenue: ${total_revenue:.2f}')

except KeyboardInterrupt:
    print('\nShutting down analytics service...')
finally:
    consumer.close()
    print('Analytics service shut down.')

What does this code do?

  • sales_count uses defaultdict(int) so any new product key automatically starts at 0. No need to check if the key exists first.
  • total_revenue accumulates every order's amount into a running total.
  • Each time a message arrives, the service updates both counters and prints the current state. This simulates a real-time dashboard that refreshes with every event.
  • group.id is analytics-group. This is the third independent consumer group, so this service also receives every order without affecting the other two.
  • Save consumer_analytics.py.
  • Test the analytics consumer by running:
python3 consumer_analytics.py
  • You should see Analytics Service started. Press Ctrl+C to stop. printed in your terminal. If existing orders are in the topic, you'll see [ANALYTICS] lines showing sales counts growing with each message.
  • Press Ctrl+C to stop the consumer.

Seeing a ModuleNotFoundError?

  • If the error mentions collections, double-check your import line says from collections import defaultdict (not collection without the 's').
  • If the error mentions confluent_kafka, your virtual environment is not active. Run source .venv/bin/activate and try again.

Help me debug my analytics consumer

✔️ Awesome, I've got everything!

Great. Make sure you've saved consumer_analytics.py and stopped the consumer before moving on.

ⓧ I'd like to double check the full code

import json
from collections import defaultdict
from confluent_kafka import Consumer

conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'analytics-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(conf)
consumer.subscribe(['orders'])

sales_count = defaultdict(int)
total_revenue = 0.0

print('Analytics Service started. Press Ctrl+C to stop.')

try:
    while True:
        msg = consumer.poll(1.0)

        if msg is None:
            continue
        if msg.error():
            print(f'Consumer error: {msg.error()}')
            continue

        order = json.loads(msg.value().decode('utf-8'))

        sales_count[order['product']] += 1
        total_revenue += order['amount']

        print(f'[ANALYTICS] Live Sales Count: {dict(sales_count)} | '
              f'Total Revenue: ${total_revenue:.2f}')

except KeyboardInterrupt:
    print('\nShutting down analytics service...')
finally:
    consumer.close()
    print('Analytics service shut down.')

You now have three independent microservices, each with its own consumer group. Next up, you'll run all four scripts simultaneously and watch the fan-out pattern come alive.

Run the Full Pipeline and Observe Fan-Out

You have four independent services ready to connect to the same Kafka topic. Now it's time to see the real power of event-driven architecture in action: running everything simultaneously and watching one stream of events fan out to multiple independent readers.

This is exactly the pattern production systems use. One team publishes order events, and downstream services (payments, fraud, analytics) each consume independently without knowing about each other.

In this step, get ready to:
  • Run all four scripts simultaneously across four terminal windows.
  • Observe offset tracking by stopping and restarting a consumer.
  • Demonstrate consumer group rebalancing with a second consumer instance.
Start all four services simultaneously

The fan-out pattern means one producer publishes events to a topic, and multiple consumers each read the full stream independently. Each consumer uses a different group.id, so Kafka treats them as separate readers.

  • Open three new tabs in Terminal by pressing Cmd+T three times.
  • In each new tab, navigate to the folder on your Desktop where your Python scripts are and activate the virtual environment by running:
source .venv/bin/activate

Why activate in every tab?

Each terminal tab starts fresh with no knowledge of the virtual environment from your original tab. Running source .venv/bin/activate tells that specific tab to use the Python packages you installed earlier (like confluent-kafka).

  • In your first terminal tab, start the producer by running:
python3 producer.py
  • In your second terminal tab, start the payments consumer by running:
python3 consumer_payments.py
  • In your third terminal tab, start the fraud detection consumer by running:
python3 consumer_fraud.py
  • In your fourth terminal tab, start the analytics consumer by running:
python3 consumer_analytics.py

All four tabs are now active. You should see the producer generating one order per second, and all three consumers processing the same events independently.

What should I see?

Each consumer tab shows every single order event the producer generates. The payments tab logs charges, the fraud tab flags or approves orders, and the analytics tab updates running totals.

This is the fan-out pattern at work. One stream of events, three completely independent readers. No consumer interferes with any other because each has its own group.id.

Not seeing output in a consumer tab?

Make sure you activated the virtual environment in that tab. Without it, Python cannot find the confluent_kafka module.

Also confirm the Kafka broker is still running with docker ps. If the container stopped, restart it with docker start kafka.

help me debug why my consumer isn't receiving messages

Demonstrate offset tracking and recovery

In production, services crash and restart all the time. Kafka tracks exactly where each consumer left off using offsets, so no events are lost during downtime.

  • Switch to the terminal tab running consumer_payments.py.
  • Press Ctrl+C to stop the payments consumer.
  • Wait 10 to 15 seconds while the producer continues generating events.
  • Restart the payments consumer by running:
python3 consumer_payments.py

Watch the payments consumer immediately process all the events it missed while it was stopped. It picks up exactly where it left off, then continues processing new events in real time.

How does offset tracking work?

Every time a consumer reads a message, Kafka records the consumer's position (called an offset) for each partition. When the consumer restarts, it asks Kafka where it last left off and resumes from there.

This is exactly how production services handle restarts, deployments, and crashes without losing events. The broker holds onto messages even after they've been read, so slow or offline consumers can always catch up.

Consumer not catching up on missed events?

If the consumer starts but only shows new events (not the missed ones), confirm that auto.offset.reset is set to earliest in your consumer_payments.py config.

Also make sure you only stopped the consumer briefly. As long as the consumer group's offsets are still committed, Kafka remembers the position.

help me understand why my consumer missed events after restarting

Observe consumer group rebalancing

When multiple instances of the same consumer (same group.id) are running, Kafka automatically distributes partitions between them. This is how production systems scale horizontally.

  • Open one more Terminal tab by pressing Cmd+T.
  • Navigate to your project folder and activate the virtual environment by running:
source .venv/bin/activate
  • Start a second instance of the payments consumer by running:
python3 consumer_payments.py

You now have two instances of consumer_payments.py running with the same group.id (payments-group).

What should I see?

Kafka automatically rebalances the three partitions between the two consumer instances. Instead of one consumer getting all events, the events are now split between the two tabs.

One instance might get events from partitions 0 and 1, while the other gets partition 2. You'll notice each tab only shows a subset of the orders. Together, they still process every event exactly once.

This is consumer group load balancing. In production, you scale a service by adding more instances with the same group ID, and Kafka handles the work distribution automatically.

  • Stop the second payments consumer instance by pressing Ctrl+C in its tab.

After stopping it, the original payments consumer resumes receiving all events from every partition. Kafka rebalanced again automatically.

Both consumers showing all events?

If both instances seem to receive every message, double-check that both are using the exact same group.id value (payments-group). Consumers with different group IDs each get a full copy of all events.

Also, rebalancing can take a few seconds. Wait about 10 seconds after starting the second instance before checking the output distribution.

help me understand Kafka consumer group rebalancing

  • Stop all running scripts by pressing Ctrl+C in each terminal tab.

What did you just demonstrate?

You ran a complete event-driven pipeline with three core Kafka capabilities:

  • Fan-out: multiple consumer groups each independently reading the full stream.
  • Offset tracking: consumers resume exactly where they left off after a restart.
  • Consumer group rebalancing: adding instances with the same group ID automatically splits the workload.

Secret mission

Dead-Letter Topic and Burst Producer

What happens when a consumer can't process a message? And how does Kafka handle a sudden flood of orders? In this mission, you'll add production-grade error handling with a dead-letter topic and stress-test your pipeline with burst traffic.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Docker container named kafka (local only, no cost).
  • Python virtual environment (.venv) with confluent-kafka installed.
  • Project files: producer.py, consumer_payments.py, consumer_fraud.py, consumer_analytics.py, consumer_fraud_dl.py, burst_producer.py.

Keep everything running

No action needed. Choose this if you're still actively experimenting with your Kafka pipeline or want to try different producer/consumer configurations.

  • Your Docker container continues running in the background.
  • Your virtual environment and project files remain on disk, ready for the next session.

Pause - I'll come back to this later

Shut down the Kafka broker to free up memory, but keep the container and its data intact. You can restart it later without any setup.

  • Stop the Kafka container by running:
docker stop kafka

Your topics, partitions, and consumer group offsets persist inside the container. When you're ready to resume, start it again by running:

docker start kafka
  • Your virtual environment and Python files stay on disk. Reactivate the virtual environment with source .venv/bin/activate when you return.

Delete - I don't want to use this again

Remove all project resources completely and start fresh.

  • Stop and remove the Kafka Docker container by running:
docker rm -f kafka
  • Deactivate your virtual environment by running:
deactivate
  • Navigate to your Desktop and delete the entire project folder by running:
cd ~/Desktop
rm -rf kafka-pipeline

This removes the virtual environment, all Python scripts, and everything else inside the project folder.

That's a Wrap!

That's a Wrap!

Nice work! You just built a real-time event streaming pipeline using Apache Kafka and Python. You now have hands-on experience with the exact architecture pattern that powers real-time data processing at scale.

You've learned how to:

  • Started a local Kafka broker in Docker using KRaft mode, created a topic with multiple partitions, and explored the broker using built-in CLI tools.
  • Built a Python producer that publishes JSON order events with key-based partitioning, ensuring orders from the same city always land in the same partition.
  • Demonstrated the fan-out pattern by running three independent consumer microservices (payments, fraud detection, analytics) each in their own consumer group, and observed offset tracking and consumer group rebalancing in action.
  • Secret Mission: Implemented a dead-letter topic pattern for failed fraud checks and stress-tested the pipeline with a burst producer sending 1000 orders in rapid succession.

Ready to quiz yourself?