Kafka Python Tutorial: From Docker to FastAPI

If you need a quick way to get Kafka running locally and talk to it from Python, this kafka python tutorial shows you how to spin up a broker with Docker, install the kafka-python client, produce messages, consume them asynchronously, and hook the flow into a FastAPI endpoint for background processing. By the end you’ll have a reproducible dev environment and a pattern you can push to production with minimal surprises.

How do I spin up a local Kafka broker with Docker?

The simplest way to start experimenting is to run a single-node Kafka cluster in Docker. The official confluentinc/cp-kafka image includes Zookeeper, so you don’t have to manage a separate service.

docker network create kafka-net

docker run -d \
  --name zookeeper \
  --network kafka-net \
  -p 2181:2181 \
  -e ZOOKEEPER_CLIENT_PORT=2181 \
  confluentinc/cp-zookeeper:7.5.0

docker run -d \
  --name kafka \
  --network kafka-net \
  -p 9092:9092 \
  -e KAFKA_BROKER_ID=1 \
  -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  confluentinc/cp-kafka:7.5.0

A few things bite me early:

• Port mapping – PLAINTEXT://localhost:9092 works for local dev but will need a proper DNS name in production.
• Resource limits – the default container uses ~1 GiB RAM. On a low-spec laptop you’ll see OOM kills unless you set -e KAFKA_HEAP_OPTS="-Xmx512M -Xms512M".
• Data persistence – without a volume, every docker rm wipes your topic data. Add -v kafka-data:/var/lib/kafka/data if you need to keep it across restarts.

Verify the broker is alive:

docker exec kafka kafka-topics --bootstrap-server localhost:9092 --list

You should see an empty list, which means the broker is reachable.

How to install and configure the kafka-python client?

For most quick prototypes kafka-python is enough. It’s pure Python, has a small footprint, and works well with FastAPI’s async model when you run the consumer in a background thread.

pip install kafka-python

Configuration is just a dict you pass to the KafkaProducer or KafkaConsumer. In production you’ll pull these values from environment variables or a config service.

KAFKA_CFG = {
    "bootstrap_servers": "localhost:9092",
    "security_protocol": "PLAINTEXT",   # switch to SASL_SSL in prod
    "client_id": "logicloop-producer",
}

A common mistake is to forget the acks setting on the producer. The default is acks=1, which can lose data if the leader crashes. For a safe dev run set:

producer = KafkaProducer(**KAFKA_CFG, acks="all", retries=5)

How to produce messages from a Python application?

Producing is straightforward. Serialize your payload to JSON (or Avro, if you have a schema registry) and send it to a topic.

import json
from kafka import KafkaProducer

producer = KafkaProducer(
    **KAFKA_CFG,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

def send_user_event(user_id: int, action: str):
    payload = {"user_id": user_id, "action": action, "ts": time.time()}
    future = producer.send("user-events", value=payload)
    try:
        record_metadata = future.get(timeout=10)
        print(f"Sent to {record_metadata.topic}:{record_metadata.partition}")
    except Exception as exc:
        print(f"Send failed: {exc}")

# Example usage
send_user_event(42, "login")
producer.flush()

What bites me in production is message ordering. Kafka only guarantees order within a partition. If you need strict ordering per user_id, hash the key to a single partition:

producer = KafkaProducer(
    **KAFKA_CFG,
    key_serializer=str.encode,
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

producer.send("user-events", key=str(user_id), value=payload)

Now all events for the same user_id land in the same partition and preserve order.

How to consume messages in Python with async support?

kafka-python does not expose an async API, but you can run the consumer inside an asyncio task that delegates to a thread pool. The pattern works fine for FastAPI background workers.

import asyncio
import json
from kafka import KafkaConsumer
from concurrent.futures import ThreadPoolExecutor

consumer = KafkaConsumer(
    "user-events",
    **KAFKA_CFG,
    group_id="logicloop-group",
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    auto_offset_reset="earliest",
)

executor = ThreadPoolExecutor(max_workers=2)

async def consume_loop():
    loop = asyncio.get_running_loop()
    while True:
        # poll blocks, so we run it in a thread
        msgs = await loop.run_in_executor(executor, consumer.poll, 1.0)
        for tp, records in msgs.items():
            for record in records:
                await handle_event(record.value)

async def handle_event(event: dict):
    # Simulate async DB write
    await asyncio.sleep(0.01)
    print(f"Processed {event}")

# Run the loop in a separate thread when the app starts
def start_consumer_background():
    asyncio.run(consume_loop())

If you prefer a truly async client, swap kafka-python for aiokafka. The API is almost identical, but you must install the extra package and adjust the startup code. I keep the kafka-python version for simplicity because it plays nicely with the existing codebase and the extra thread overhead is negligible for low-throughput services.

When I first tried to run the consumer directly in the FastAPI event loop, the app hung on startup. The fix was to launch the consumer in a separate daemon thread (or use lifespan events). See the FastAPI async pitfalls in the internal link Fixing SQLAlchemy MissingGreenlet Error in FastAPI (Async Explained) for more context on mixing sync and async code.

How can I integrate Kafka with FastAPI for background processing?

FastAPI gives you two natural hooks: startup events to launch the consumer and background tasks to fire-and-forget work that originates from HTTP requests.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

@app.on_event("startup")
async def start_kafka_consumer():
    loop = asyncio.get_event_loop()
    loop.create_task(consume_loop())   # consume_loop from previous section

@app.post("/events/")
async def publish_event(user_id: int, action: str, background: BackgroundTasks):
    send_user_event(user_id, action)   # sync producer, cheap call
    # Optionally schedule extra processing
    background.add_task(log_audit, user_id, action)
    return {"status": "queued"}

def log_audit(user_id: int, action: str):
    # This runs in a thread pool managed by FastAPI
    time.sleep(0.1)  # simulate I/O
    print(f"Audit log for {user_id}:{action}")

A few production-grade concerns:

• Graceful shutdown – FastAPI’s shutdown event should close the consumer and the thread pool to avoid “dangling thread” warnings.
• Back-pressure – If the consumer can’t keep up, messages accumulate in the broker. Tune max_poll_records and consider scaling the consumer group horizontally.
• Observability – Export consumer lag metrics to Prometheus. The kafka-python client exposes consumer.position(tp) and consumer.end_offsets([tp]) which you can compare.

When I first deployed this pattern to Google Cloud Run, the container was killed after 15 minutes of idle time, causing the consumer to miss messages. The fix was to add a tiny “keep-alive” endpoint that Cloud Run pings, and to set the --timeout flag higher. The full deployment story is covered in Serverless Python: Deploying FastAPI to Google Cloud Run with Docker.

Common pitfalls and best-practice tips

When you move beyond a single node, you’ll need to think about replication factor, multi-region latency, and security (SASL/SCRAM, TLS). Those concerns are out of scope for this tutorial but worth planning early.

FAQ

What’s the difference between kafka-python and aiokafka?
kafka-python is a synchronous client that works well with a thread-based approach. aiokafka is built on asyncio and offers true non-blocking APIs, but adds an extra dependency and a slightly steeper learning curve.

Do I need Zookeeper for a production Kafka cluster?
Modern Kafka versions can run without an external Zookeeper when you enable KRaft mode. The Docker images used in this tutorial still rely on Zookeeper for simplicity, but for new clusters consider KRaft to reduce operational overhead.

How many partitions should I create for a topic?
Start with a number that matches your expected consumer parallelism (e.g., 3-6 partitions for a small service). You can increase partitions later, but you cannot decrease them without recreating the topic.

Can I use this setup on Windows?
Docker Desktop runs Linux containers on Windows, so the broker part works the same. The only Windows-specific issue is the path handling in volume mounts; use absolute Unix-style paths inside the container.

Key Takeaways

• Docker lets you spin up a single-node Kafka broker in seconds; remember to persist data if you need it across restarts.
• kafka-python is a lightweight client; configure acks, retries, and proper serialization to avoid data loss.
• Produce messages with a key if you need ordering per entity.
• Consume asynchronously by delegating the blocking poll to a thread pool; for pure async workloads consider aiokafka.
• FastAPI startup/shutdown events are the right place to launch and stop a Kafka consumer; background tasks handle fire-and-forget work.
• Watch for common pitfalls: offset management, back-pressure, resource limits, and security settings.

With these pieces in place you have a solid foundation for building reliable, event-driven services in Python. The next step is to push the Docker image to your CI/CD pipeline - see the guide on Automating Production: A CI/CD Pipeline for Google Cloud Run with GitHub Actions for a production-ready workflow.