Event-Driven Order Pipeline with Kafka

Build a local order pipeline with Kafka, Docker, and Flask.

Introduction

30 Second Summary

When you place an order online, your request travels through multiple systems. Inventory, payment, and shipping each handle their part of the job. These systems don't call each other directly. Instead, they pass messages through a central broker that routes events to the right place.

In this project, you will build a local event-driven order pipeline using Apache Kafka and Docker. A Flask API publishes order events to a Kafka topic, and a separate consumer script processes them in real time.

What You'll Build

A minimal event-driven order processing system where a REST API sends order messages through a Kafka broker to a consumer that logs and commits each one.

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

  • A running Kafka broker in Docker that you launched with a single Docker Compose command.
  • A Flask REST API that publishes JSON order messages to a Kafka topic.
  • A consumer script that polls the topic in a consumer group and manually commits offsets.
  • Secret Mission: add a dead letter topic that captures malformed messages so bad data never blocks the pipeline.

Are there any prerequisites?

You need Docker Desktop and Python 3.14.6 installed on your machine before starting. Basic Python knowledge (variables, functions, dictionaries) is helpful but you don't need prior experience with Kafka or Docker Compose.

Set Up Your Environment

Before you can build the event-driven pipeline, you need a workspace with the right tools ready to go.

In this step, you'll install Cursor as your code editor, create a Python environment with the libraries that talk to Kafka, and confirm Docker is ready to run containers.

In this step, get ready to:
  • Install Cursor and create the project folder.
  • Create a Python virtual environment and install dependencies.
  • Verify Docker Desktop is running.
Install Cursor and create the project folder

Cursor is an AI-powered code editor that you'll use to write and manage all the project files.

  • Open your browser and go to cursor.com/download.
  • Click the download button for your operating system (macOS or Windows).
  • Run the downloaded installer.

macOS

  • Drag Cursor into your Applications folder.

Windows

  • Follow the installer prompts and click Next until installation completes.
  • When prompted, sign up for a free Cursor account or sign in if you already have one.

Cursor won't open?

On macOS, you may see a security prompt the first time. Click Open to allow it. On Windows, if SmartScreen blocks the installer, click More info then Run anyway.

Help me troubleshoot Cursor installation

Now you'll create a project folder on your Desktop and open it in Cursor.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.

You need a terminal inside Cursor to run commands. Open one now.

  • Open the integrated terminal by pressing Ctrl+` (backtick key, top-left of your keyboard) on both macOS and Windows.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir kafka-orders
cd kafka-orders
  • Confirm the folder exists by running:
ls

You should see an empty output since the folder is brand new.

  • Open this folder as a workspace in Cursor by clicking File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the kafka-orders folder, and click Open.

Why open the folder as a workspace?

Opening the folder tells Cursor to treat it as your project root. The file explorer in the sidebar will show only your project files, and the terminal will start inside the correct directory.

Create a virtual environment and install dependencies

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

  • In the Cursor terminal, create a virtual environment by running this command:

macOS

python3 -m venv .venv

Windows

python -m venv .venv

What does this command do?

The -m venv flag tells Python to run its built-in venv module. .venv is the name of the folder where the isolated environment lives. The dot prefix keeps it hidden in file browsers.

  • Activate the virtual environment by running:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate

You should see (.venv) appear at the beginning of your terminal prompt. This confirms the environment is active.

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

On Windows, if you see a permissions error, you may need to allow script execution. Run Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned in PowerShell, then try the activate command again.

On macOS, make sure you typed source before the path. Without it, the shell tries to execute the file directly and fails.

Help me activate my virtual environment

  • Install Flask and the confluent-kafka library by running:
pip install flask==3.1.3 confluent-kafka==2.14.2

What are these packages?

Flask is a lightweight web framework that you'll use to build the HTTP endpoint that accepts orders. The confluent-kafka library is a high-performance Python client for Apache Kafka, built on top of the C library librdkafka.

Pinning exact versions (==3.1.3 and ==2.14.2) ensures everyone gets the same behaviour regardless of when they run the install.

The install output should end with a line showing both packages were successfully installed.

Install failed with an error?

Make sure (.venv) appears in your prompt. If you forgot to activate the virtual environment, pip might try to install globally and fail due to permissions.

If you see a network timeout, check your internet connection and try running the command again.

Help me fix my pip install error

Verify Docker is running

The Kafka broker will run inside a Docker container. You need to confirm Docker Desktop is running and that Docker Compose is available.

  • Check that Docker is available by running this command:
docker --version
  • Check that Docker Compose is available by running:
docker compose version

Both commands should print a version number. If they do, Docker Desktop is running and ready.

Command not found or connection error?

If you see a connection error or "Cannot connect to the Docker daemon", Docker Desktop is not running. Open Docker Desktop from your applications menu and wait for it to fully start (the whale icon in your system tray will stop animating).

If docker compose version fails but docker --version works, make sure you have a recent version of Docker Desktop. Compose V2 is bundled with Docker Desktop by default.

Help me troubleshoot Docker

Your workspace is ready. Next up, you'll launch a Kafka broker inside Docker with a single compose file.

Launch Kafka with Docker Compose

Your development environment is ready. Now it's time to bring up the infrastructure your event-driven system depends on.

In this step, you'll launch a Kafka broker inside a Docker container using Docker Compose. You'll also create the topic that your producer and consumer will communicate through.

In this step, get ready to:
  • Write a docker-compose.yml that configures a single Kafka broker in KRaft mode.
  • Start the Kafka container and verify the broker is reachable.
  • Create the orders topic that your producer will publish messages to.
Write the docker-compose.yml

Docker Compose lets you define and run multi-container applications with a single configuration file. Your Kafka broker will run in KRaft mode, which means Kafka handles its own metadata internally. No separate ZooKeeper container needed.

  • In Cursor, create a new file called docker-compose.yml in the root of your kafka-orders folder.
  • Paste the following configuration into the file:
services:
  kafka:
    image: apache/kafka:4.3.1
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

What does this configuration do?

  • image: apache/kafka:4.3.1 pulls the official Apache Kafka Docker image at version 4.3.1.
  • KAFKA_PROCESS_ROLES: broker,controller tells this single node to act as both the message broker and the metadata controller (KRaft mode).
  • KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 is the address Kafka tells your Python app to connect to. Since your app runs natively on your machine (not inside Docker), localhost is correct.
  • The replication factors are all set to 1 because you only have a single broker. Kafka's default is 3, which would fail on a single-node cluster.

Why KRaft mode instead of ZooKeeper?

Older versions of Kafka required a separate ZooKeeper service to manage cluster metadata. KRaft eliminates that dependency entirely. Your setup needs just one container instead of two, making it simpler to run and reason about.

  • Save docker-compose.yml.

File not saving?

Make sure the file is named exactly docker-compose.yml (not docker-compose.yaml or dockercompose.yml). It must be in the root of your kafka-orders folder, not inside a subfolder.

Help me fix my docker-compose.yml file location

✔️ 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

services:
  kafka:
    image: apache/kafka:4.3.1
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
Start Kafka and verify

With the configuration saved, you can now launch the Kafka container. Docker Compose reads your file and handles the rest.

  • Open a terminal in Cursor (press Cmd+` on macOS or Ctrl+` on Windows).
  • Start the Kafka container in the background by running:
docker compose up -d
  • Check the container status by running:
docker compose ps

You should see the kafka container listed with a status showing Up. This confirms the container started successfully.

  • Verify the broker is actually accepting connections by running:
docker exec kafka /opt/kafka/bin/kafka-cluster.sh cluster-id --bootstrap-server localhost:9092

You should see a cluster ID printed (a string of characters like MkU3OEVBNTcwNTJENDM2Qg). This confirms the Kafka broker is running and reachable.

Container not starting or broker not reachable?

  • Make sure Docker Desktop is running. Check for its icon in your system tray or menu bar.
  • If you see a port conflict error, another service may be using port 9092. Stop other containers with docker compose down in any other project folders, then try again.
  • If the cluster-id command fails, wait 10-15 seconds and try again. Kafka needs a moment to finish its startup sequence.

Help me troubleshoot my Kafka container startup

Create the orders topic

A Kafka topic is a named channel that messages flow through. Your producer will write to a topic called orders, and your consumer will read from it.

  • Create the orders topic by running:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --create --topic orders --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
  • Verify the topic was created by running:
docker exec kafka /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

You should see orders in the output.

What is a Kafka topic?

Think of a topic as a named log file that Kafka manages for you. Producers append messages to the end of the log. Consumers read from the log at their own pace, tracked by an offset (a position number).

The --partitions 1 flag means all messages go into a single ordered stream. In production, you'd use multiple partitions to allow parallel consumers. For learning the fundamentals, one partition keeps things simple.

Topic creation failed?

  • If you see a connection error, the broker may not have finished starting. Wait a few seconds and retry the create command.
  • If the topic already exists (from a previous attempt), the --list command will still show it. You can safely continue.

Help me fix my Kafka topic creation

Your Kafka infrastructure is live. You have a running broker and a topic ready to receive messages. Next up, you'll build the Flask producer that publishes orders into this topic.

Build the Flask Producer

Your Kafka broker is running and the orders topic is ready to receive messages. Now you need a way to get data into that topic.

In this step, you'll create a Flask API that accepts order data over HTTP and publishes each order as a message to Kafka. This is the producer side of the event-driven pipeline.

In this step, get ready to:
  • Create producer.py with a Flask app and Kafka producer.
  • Add a delivery report callback to confirm message delivery.
  • Build a POST /order endpoint that publishes orders to Kafka.
Write producer.py

The producer script needs two things: a Flask web server to receive HTTP requests, and a Kafka producer client to forward those requests as messages.

  • In Cursor, create a new file called producer.py inside your kafka-orders folder.
  • Add the imports and initialization code by pasting this into the file:
from flask import Flask, request, jsonify
from confluent_kafka import Producer
import json

app = Flask(__name__)

producer = Producer({'bootstrap.servers': 'localhost:9092'})

What does this code do?

  • The imports pull in Flask for the web server and confluent_kafka for the Kafka client.
  • app = Flask(__name__) creates a Flask application instance.
  • The Producer connects to your Kafka broker at localhost:9092. This is the same port your Docker container exposes.
  • Confirm the file exists by running:
ls producer.py

You should see producer.py listed in the output.

File not found?

  • Make sure you saved the file in Cursor before running ls.
  • Check that your terminal is in the kafka-orders folder. Run pwd to confirm.

Why isn't my producer.py file showing up?

  • Next, add the delivery report callback below the producer initialization in producer.py:
def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")
    else:
        print(f"Message delivered to {msg.topic()} [{msg.partition()}] at offset {msg.offset()}")

What does the delivery report do?

Every time Kafka finishes processing a produced message, it calls this function. If delivery failed, you see the error. If it succeeded, you see the exact topic, partition, and offset where the message landed.

This feedback loop is essential for debugging. Without it, messages could silently fail.

  • Save producer.py.

You won't see a visible change yet. The next chunk adds the endpoint that uses this callback.

Add the POST /order endpoint

The endpoint is where HTTP meets Kafka. It receives order data from a web request, packages it as a JSON message, and hands it off to the broker.

  • Add the following route and main block below the delivery_report function in producer.py:
@app.route('/order', methods=['POST'])
def create_order():
    data = request.get_json()
    order = {
        'order_id': data['order_id'],
        'item': data['item'],
        'quantity': data['quantity']
    }
    producer.produce(
        'orders',
        key=str(order['order_id']),
        value=json.dumps(order),
        callback=delivery_report
    )
    producer.flush()
    return jsonify({'status': 'order published', 'order': order}), 201


if __name__ == '__main__':
    app.run(debug=True, port=5000)

What does this endpoint do?

  • request.get_json() extracts the JSON body from the incoming HTTP request.
  • producer.produce() sends the order to the orders topic, using the order_id as the message key and the full order dict as the value.
  • producer.flush() blocks until Kafka confirms delivery. This ensures the HTTP response only goes back after the message is safely stored.
  • The function returns a 201 status code, meaning "created successfully."
  • The if __name__ block starts Flask on port 5000 when you run the script directly.
  • Save producer.py.
  • Verify the file runs without errors by starting Flask:
python producer.py

You should see output including Running on http://127.0.0.1:5000. This confirms your producer is ready to accept requests.

  • Press Ctrl+C to stop the server for now. You'll start it again in the testing step.

Seeing an import error?

  • Make sure your virtual environment is activated. You should see (.venv) at the beginning of your terminal prompt.
  • If Flask or confluent-kafka is missing, run pip install flask==3.1.3 confluent-kafka==2.14.2 again.
  • Check for typos in your import lines. The package is confluent_kafka with an underscore, not a hyphen.

Why is my Flask producer failing to start?

✔️ Awesome, I've got everything!

Great. Double check you've saved producer.py and you're ready for the next step.

ⓧ I'd like to double check the full code

Here's the complete producer.py file for comparison:

from flask import Flask, request, jsonify
from confluent_kafka import Producer
import json

app = Flask(__name__)

producer = Producer({'bootstrap.servers': 'localhost:9092'})


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


@app.route('/order', methods=['POST'])
def create_order():
    data = request.get_json()
    order = {
        'order_id': data['order_id'],
        'item': data['item'],
        'quantity': data['quantity']
    }
    producer.produce(
        'orders',
        key=str(order['order_id']),
        value=json.dumps(order),
        callback=delivery_report
    )
    producer.flush()
    return jsonify({'status': 'order published', 'order': order}), 201


if __name__ == '__main__':
    app.run(debug=True, port=5000)

Your producer is built and confirmed working. Next up, you'll build the consumer that reads messages from the orders topic and processes them.

Build the Consumer

Your producer is ready to publish order events to Kafka. Now you need something on the other side to pick those messages up and act on them.

In this step, you'll build a consumer script that connects to the same orders topic, reads messages as they arrive, and commits its position so it never reprocesses the same order twice.

In this step, get ready to:
  • Write a consumer script that connects to your Kafka broker and subscribes to the orders topic.
  • Implement a poll loop that deserializes messages and manually commits offsets.
Write consumer.py

The consumer is a separate script that runs independently from your Flask producer. It connects to Kafka as part of a consumer group, which lets Kafka track how far this consumer has read through the topic.

  • Create a new file called consumer.py in your kafka-orders folder by pasting the following code:
from confluent_kafka import Consumer, KafkaError
import json

# Configure the consumer with broker address and group settings
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': 'false'
})

# Subscribe to the orders topic
consumer.subscribe(['orders'])

print("Consumer started. Waiting for orders...")

What does this code do?

This sets up your Kafka consumer with four key configuration values:

  • bootstrap.servers tells the consumer where to find the Kafka broker (the same address your producer uses).
  • group.id assigns this consumer to the order-processors group. Kafka uses this to track which messages have been read.
  • auto.offset.reset set to earliest means the consumer starts reading from the beginning of the topic if it has no saved position.
  • enable.auto.commit set to false means you control exactly when offsets are saved. This prevents marking a message as processed before you've actually handled it.
  • Save consumer.py.
  • Run the script to confirm it connects successfully by running:
python consumer.py

You should see Consumer started. Waiting for orders... print before the script exits. This confirms your consumer can reach the broker and subscribe to the topic.

Seeing a connection error?

  • Make sure your Kafka container is still running. Check with docker compose ps and confirm the kafka container shows as running.
  • Verify your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.

Help me debug a connection error in my Kafka consumer script

Implement the poll loop

The consumer needs to continuously check for new messages. This is called polling. Your poll loop will wait up to one second for a message, process it if one arrives, then repeat.

  • In consumer.py, add the following code below the print("Consumer started. Waiting for orders...") line:
try:
    while True:
        # Poll for a message with a 1-second timeout
        msg = consumer.poll(1.0)

        # No message available yet
        if msg is None:
            continue

        # Handle any errors from the broker
        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f"Consumer error: {msg.error()}")
                break

        # Deserialize and process the order
        order = json.loads(msg.value().decode('utf-8'))
        print(f"Processing order: {order}")
        print(f"  Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset()}")

        # Manually commit the offset after successful processing
        consumer.commit()
        print(f"  Offset committed.")

except KeyboardInterrupt:
    print("Shutting down consumer...")
finally:
    # Always close the consumer to leave the group cleanly
    consumer.close()

What does this code do?

This is the heart of your consumer. Here is what each section handles:

  • consumer.poll(1.0) waits up to 1 second for a new message. If nothing arrives, it returns None and the loop continues.
  • The error check handles edge cases like reaching the end of a partition (KafkaError._PARTITION_EOF), which is normal and just means there are no more messages right now.
  • json.loads(msg.value().decode('utf-8')) converts the raw bytes back into a Python dictionary.
  • consumer.commit() saves the current offset AFTER processing. If the consumer crashes before committing, it will re-read and re-process that message on restart.
  • consumer.close() in the finally block ensures the consumer leaves the group cleanly when you press Ctrl+C.

Why commit manually instead of automatically?

With auto-commit enabled, Kafka marks a message as processed the moment it is read. If your code crashes between reading and actually handling the message, that order is lost forever.

Manual commits give you a guarantee: the offset only advances after your code successfully processes the message. This is critical in real-world systems where losing an order means losing revenue.

  • Save consumer.py.
  • Run your consumer by running:
python consumer.py

You should see Consumer started. Waiting for orders... and the script keeps running, waiting for messages. It will not exit this time because the poll loop runs continuously.

  • Press Ctrl+C to stop the consumer for now. You should see Shutting down consumer... print as it exits cleanly.

Consumer exits immediately without waiting?

  • Make sure the try: block starts directly after the print statement. If the poll loop code is outside the try block, an unhandled error could cause immediate exit.
  • Check your indentation. The while True: loop must be indented inside try: for the KeyboardInterrupt handling to work.

Help me debug why my Kafka consumer exits immediately instead of polling

✔️ Awesome, I've got everything!

Great. Double check you have saved consumer.py and move on to the next step.

ⓧ I'd like to double check the full code

from confluent_kafka import Consumer, KafkaError
import json

consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'order-processors',
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': 'false'
})

consumer.subscribe(['orders'])

print("Consumer started. Waiting for orders...")

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

        if msg is None:
            continue

        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                continue
            else:
                print(f"Consumer error: {msg.error()}")
                break

        order = json.loads(msg.value().decode('utf-8'))
        print(f"Processing order: {order}")
        print(f"  Topic: {msg.topic()}, Partition: {msg.partition()}, Offset: {msg.offset()}")

        consumer.commit()
        print(f"  Offset committed.")

except KeyboardInterrupt:
    print("Shutting down consumer...")
finally:
    consumer.close()

Your consumer is ready to process orders. Next up, you'll run both scripts side by side and watch messages flow from the producer through Kafka to the consumer in real time.

Run and Test End-to-End

Your producer and consumer are built. Now it's time to fire them both up and watch messages flow through Kafka in real time.

You'll run three terminals side-by-side: one for the Flask producer, one for the consumer, and one to send HTTP requests. This lets you see how a message travels from an API call, into a Kafka topic, and out the other side to a consumer.

In this step, get ready to:
  • Start the Flask producer and consumer in separate terminals.
  • Send test orders via HTTP and observe them flowing through Kafka.
  • Verify that committed offsets persist across consumer restarts.
Start the producer and consumer

You need two separate terminal processes running at the same time. One runs Flask (the producer), and the other runs the consumer poll loop.

  • In Cursor, open a new terminal by pressing Ctrl+Shift+` (macOS) or Ctrl+Shift+` (Windows).
  • This will be Terminal 1, where you run the producer.
  • Activate your virtual environment in Terminal 1 by running:

macOS

source .venv/bin/activate

Windows

.venv\Scripts\activate
  • Start the Flask producer by running:
python producer.py

You should see Flask start up with output like Running on http://127.0.0.1:5000.

  • Open a second terminal by pressing Ctrl+Shift+` again.
  • This will be Terminal 2, where you run the consumer.
  • Activate the virtual environment in Terminal 2 using the same command as above.
  • Start the consumer by running:
python consumer.py

You should see Consumer started. Waiting for orders... printed in Terminal 2.

Flask won't start or consumer errors out?

If you see an import error, make sure you activated the virtual environment in that terminal. The (.venv) prefix should appear at the start of your terminal prompt.

If the consumer shows a connection error, confirm your Kafka container is still running with docker compose ps. If it stopped, run docker compose up -d to restart it.

Still stuck? Help me troubleshoot why my producer or consumer won't start.

Send test orders and observe the flow

With both processes running, you can send HTTP requests to the Flask producer and watch messages appear in the consumer. This proves the full pipeline works: HTTP request to producer to Kafka topic to consumer.

  • Open a third terminal by pressing Ctrl+Shift+` once more.
  • Send your first test order by running this command:

macOS

curl -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 1, "item": "keyboard", "quantity": 2}'

Windows

curl.exe -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 1, "item": "keyboard", "quantity": 2}'

What should I see?

Three things happen across your terminals:

  • Terminal 3 (curl): A JSON response with "status": "order published" and the order data.
  • Terminal 1 (producer): A delivery confirmation like Message delivered to orders [0] at offset 0.
  • Terminal 2 (consumer): The processed order with topic, partition, and offset details, followed by Offset committed..
  • Send a few more orders to see offsets incrementing. Run these commands one at a time:

macOS

curl -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 2, "item": "mouse", "quantity": 1}'

curl -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 3, "item": "monitor", "quantity": 1}'

Windows

curl.exe -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 2, "item": "mouse", "quantity": 1}'

curl.exe -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 3, "item": "monitor", "quantity": 1}'

In Terminal 2, each new order should appear with an incrementing offset (0, 1, 2). This confirms Kafka is tracking message positions within the topic.

Not seeing output in the consumer?

Make sure Terminal 2 shows Consumer started. Waiting for orders... and is still running. If it exited, restart it with python consumer.py.

If curl returns a connection error, check that Terminal 1 still shows Flask running on port 5000.

Need help? My curl command runs but the consumer doesn't print anything.

Verify offset persistence

The whole point of committing offsets is that Kafka remembers where a consumer left off. If the consumer crashes and restarts, it should pick up from the last committed position, not replay every message from the beginning.

  • Switch to Terminal 2 (the consumer).
  • Stop the consumer by pressing Ctrl+C.

You should see Shutting down consumer... as the graceful shutdown triggers.

  • While the consumer is stopped, send another order from Terminal 3:

macOS

curl -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 4, "item": "headphones", "quantity": 1}'

Windows

curl.exe -X POST http://localhost:5000/order -H "Content-Type: application/json" -d '{"order_id": 4, "item": "headphones", "quantity": 1}'

The producer still publishes to Kafka even though no consumer is listening. The message sits in the topic, waiting.

  • Switch back to Terminal 2 and restart the consumer by running:
python consumer.py

What should I see?

The consumer picks up ONLY the new headphones order (offset 3). It does not replay the keyboard, mouse, or monitor orders you sent earlier.

This proves that your consumer.commit() calls saved the consumer group's position to the Kafka broker. When the consumer restarted and rejoined the order-processors group, Kafka told it exactly where to resume.

Why does offset persistence matter?

In production, consumers crash, restart for deployments, or scale horizontally. Without committed offsets, every restart would reprocess the entire topic from the beginning. Manual commits give you control over when Kafka considers a message "done."

Consumer replays old messages on restart?

If you see all previous orders again, check that enable.auto.commit is set to 'false' (as a string) in your consumer config, and that consumer.commit() is called after processing each message.

Also confirm that group.id is the same value (order-processors) every time the consumer starts. A different group ID means Kafka treats it as a brand-new consumer with no committed offsets.

Need a hand? My consumer replays old messages after restarting instead of picking up where it left off.

Secret mission

Add a Dead Letter Topic

In production Kafka pipelines, malformed messages can crash your consumer or block the entire topic. Your mission: add a dead letter topic that catches bad messages and lets the consumer keep processing valid ones.

Clean Up Your Resources

Clean Up Your Resources

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

Resources you used:

  • One Docker container running Kafka (uses minimal CPU and memory when idle).
  • Local project folder kafka-orders containing your Python files and virtual environment.

Keep everything running

No action needed. Choose this if you're still actively building or want to experiment with Kafka further.

  • Leave Docker Desktop running. The Kafka container stays available for future experiments.
  • You can send more orders or create new topics whenever you want.

Pause - I'll come back to this later

Shut down the Kafka container to free up memory, but keep all your files and container data intact.

  • Open a terminal in your kafka-orders folder.
  • Stop the container by running:
docker compose stop
  • When you're ready to come back, restart the container by running:
docker compose start
  • Your topics, offsets, and messages are preserved.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes the container, network, and all local files.

  • Open a terminal in your kafka-orders folder.
  • Remove the Kafka container and its network by running:
docker compose down
  • Move up one directory and delete the project folder.

On macOS or Linux:

cd ..
rm -rf kafka-orders

On Windows:

cd ..
rmdir /s /q kafka-orders
  • Confirm the folder is gone by running ls (macOS/Linux) or dir (Windows). You should no longer see kafka-orders in the output.

That's a Wrap!

That's a Wrap!

Nice work! You just built a fully functional event-driven order pipeline using Apache Kafka and Docker.

You've learned how to:

  • Set up a Kafka broker in Docker using KRaft mode with a single docker-compose.yml.
  • Built a Flask REST API that publishes JSON messages to a Kafka topic and verified delivery with a callback.
  • Built a consumer script that polls a topic in a consumer group with manual offset commits and proved offset persistence end-to-end.
  • Secret Mission: Added a dead letter topic for routing malformed messages with error headers.

Ready to quiz yourself?