Kafka in production in Python: aiokafka, DLQ, Schema Registry, tuning, security

← Back to the section

If you already know what a topic, a partition, and a consumer group are, this article is about the next step: how to connect Kafka to a Python application and what you need to configure so that it runs reliably in production.

aiokafka: producer and consumer loop

The aiokafka library is an asynchronous Kafka client for asyncio. To send messages you use AIOKafkaProducer; to receive them you use AIOKafkaConsumer with a consumer loop.

# Send
class OrderEventPublisher:
    def __init__(self, producer: AIOKafkaProducer):
        self._producer = producer

    async def publish(self, event: OrderEvent) -> None:
        await self._producer.send_and_wait(
            "orders",
            key=str(event.order_id).encode(),
            value=serialize(event),
        )

# Receive
async def order_event_listener() -> None:
    consumer = AIOKafkaConsumer("orders", group_id="billing-service")
    await consumer.start()
    try:
        async for record in consumer:
            event = deserialize(record.value)
            # processing
    finally:
        await consumer.stop()

Parallelism means several consumer tasks in the same group: each creates its own AIOKafkaConsumer and serves its own set of partitions. By default there is one task, but you can start more:

tasks = [asyncio.create_task(order_event_listener()) for _ in range(3)]

Commit offset: when Kafka "knows" a message has been processed

Kafka tracks how far a consumer has progressed through the messages via an offset commit. When that commit happens is up to you:

Auto-commit — every auto_commit_interval_ms (5 seconds, the default value), regardless of whether the message has been processed.
After a batch — getmany() + commit() after the whole batch has been processed successfully. Faster, but on a failure the whole batch is re-read.
After each message — commit() inside the loop body. Safer on failures, but slower. Needed when it matters to commit the offset only after a successful write to the database.

consumer = AIOKafkaConsumer(
    "orders",
    group_id="billing-service",
    enable_auto_commit=False,
)
await consumer.start()
try:
    while True:
        batches = await consumer.getmany(timeout_ms=1000, max_records=100)
        for tp, records in batches.items():
            for record in records:
                await handle(record)
        await consumer.commit()  # after the whole batch has been processed successfully
finally:
    await consumer.stop()

Message headers

Every Kafka message can carry headers — (key, value) pairs in bytes, separate from the payload. This is where you put technical metadata that does not belong in the business data:

X-Correlation-ID / traceparent — for distributed tracing.
X-Event-Version — the version of the event format.
X-Source-Service — where the message came from.

# When sending
await producer.send_and_wait(
    "orders",
    key=key,
    value=serialize(event),
    headers=[("X-Correlation-ID", correlation_id.encode())],
)

# When receiving
async for record in consumer:
    headers = dict(record.headers)
    correlation_id_var.set(headers["X-Correlation-ID"].decode())  # contextvars for logging
    # processing

Business data (orderId, amount, status) goes in the payload. Everything technical goes in the headers.

Dead Letter Queue: what to do with problematic messages

If a handler throws an exception and the offset is not committed, after a restart the consumer keeps receiving the same message over and over — the offset does not move and the consumer stalls. In production this means the whole group grinds to a halt.

The solution is a Dead Letter Queue (DLQ): after several failed attempts the message is moved to a separate topic (usually with a .DLT suffix), the offset is committed, and the main consumer keeps working.

The simple way: retries with a DLQ in the handler

MAX_ATTEMPTS = 3

async def handle_with_dlq(record: ConsumerRecord) -> None:
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            await handle(record)
            return
        except Exception as e:
            if attempt == MAX_ATTEMPTS:
                await producer.send_and_wait(
                    "orders.DLT",
                    key=record.key,
                    value=record.value,
                    headers=[
                        ("x-exception-message", str(e).encode()),
                        ("x-original-topic", record.topic.encode()),
                    ],
                )
            else:
                await asyncio.sleep(1.0)

Here: 3 attempts with a 1-second interval, and after the third failure the message goes to orders.DLT. The DLT message gets headers with the error text and the name of the original topic.

The non-blocking way: retry topics

Retries inside the handler block the partition: while one message "cools off", the rest are waiting. The alternative is separate retry topics with an increasing delay:

# orders → orders-retry-0 (1s) → orders-retry-1 (2s) → orders-retry-2 (4s) → orders-dlt
async def handle_or_escalate(record: ConsumerRecord) -> None:
    try:
        await handle(deserialize(record.value))
    except Exception as e:
        next_topic = next_retry_topic(record.topic)
        await producer.send_and_wait(
            next_topic,
            key=record.key,
            value=record.value,
            headers=[("x-exception-message", str(e).encode())],
        )

async def retry_consumer(topic: str, delay: float) -> None:
    async for record in retry_topic_consumer(topic):
        await asyncio.sleep(delay)  # the message "cools off" without blocking the main topic
        await handle_or_escalate(record)

Each retry topic gets its own consumer with its own delay. Messages "cool off" in the retry topics without blocking the main flow.

When to use which

Temporary problem (database down, external service not responding) → retry topics with an increasing delay.
Permanent problem (invalid message, wrong format) → no retries, straight to the DLQ.
Need to distinguish error types → check the exception type: ValueError (invalid data) goes straight to the DLQ, ConnectionError gets retried.

Schema Registry: how not to break your neighbors when the schema changes

When the producer and the consumer are different services from different teams, any change to the event structure can potentially break the consumer. Schema Registry solves this with a centralized schema store that checks compatibility.

How it works:

The producer registers the schema in Schema Registry and gets a numeric identifier (schema_id).
Into each message it writes 4 bytes with that identifier, followed by the compressed payload.
The consumer reads the schema_id, downloads the schema from Schema Registry once (then caches it), and deserializes the payload.

Avro, Protobuf, or JSON Schema

The standard in the Kafka ecosystem is Avro: a compact binary format with good support for schema evolution via confluent-kafka with fastavro. Protobuf is chosen by teams with gRPC infrastructure. JSON Schema is human-readable but takes several times more space — it is good for debugging.

Compatibility modes

Schema Registry checks whether a new version of the schema will break consumers that are already running:

BACKWARD (the default) — a new consumer can read old messages. You can remove optional fields and add optional ones with a default value. You cannot add required fields.
FORWARD — an old consumer can read new messages. The mirror image of BACKWARD.
FULL — both modes at once. The strictest.
NONE — no checks. Only for special cases.

Rule of thumb: BACKWARD for topics where there are many consumers and a single producer — that is the typical event bus.

Consumer lag: why the consumer falls behind

Lag is the difference between the last message in a partition and the one the consumer has reached. If lag is growing, the consumer cannot keep up with the producer. This is the main health metric for a consumer in production.

Typical causes:

Slow processing — each message makes a synchronous call to a database or an external service. Fix: reduce max_poll_records, parallelize processing, switch to batch requests.
Too few partitions per group — add consumer tasks or service instances (but no more than the number of partitions).
max_poll_interval_ms expires — if processing one batch takes longer than 5 minutes (the default value), Kafka considers the consumer dead and triggers a partition rebalance. Fix: reduce max_poll_records or raise max_poll_interval_ms.
Blocking code in the event loop — a synchronous call froze for 30 seconds, stopped all the consumer tasks, and lag grew by thousands of messages.

Check the lag manually:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group billing-service --describe
# TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders  0          124530          124530          0
# orders  1          124450          124530          80   ← falling behind

Metric to monitor: kafka_consumergroup_lag. Alert: lag is above N and has been growing for N minutes in a row.

Performance tuning

Kafka works well out of the box, but there are a few parameters worth knowing.

Producer

Parameter	Default value	When to change
`max_batch_size`	16 KB	Raise to 64–256 KB under high write volume. The producer waits until a batch fills up and sends it as a whole.
`linger_ms`	0	Raise to 5–20 ms. The producer waits N ms before sending, even if the batch is not full — more throughput, slightly more latency.
`compression_type`	`None`	Enable `zstd` or `lz4`. On JSON messages it saves 3–5x on traffic and disk space.

Consumer

Parameter	Default value	When to change
`fetch_min_bytes`	1 byte	Raise to 50–500 KB. The consumer waits for data to accumulate — less load on the broker.
`max_poll_records`	500	Lower to 100–50 if each message requires long processing.
`max_poll_interval_ms`	5 minutes	Raise if processing a batch really does take that long.

The best compression algorithm for most cases is zstd (Kafka 2.1+): it compresses almost as well as gzip and runs almost as fast as lz4.

Security: SASL, SSL, ACL

By default Kafka listens on an open port with no authentication — you never leave it that way in production. Three layers of protection:

SSL/TLS — connection encryption. Certificates on the brokers and clients.
SASL — authentication. SASL/PLAIN — username and password (only over SSL). SASL/SCRAM-SHA-256 — safer: challenge-response, the password is not sent over the network. SASL/OAUTHBEARER — OAuth2 tokens, for integration with an identity provider.
ACL — authorization: who is allowed to read from and write to which topic. Managed via kafka-acls.sh.

Typical setup: one SASL user per service, with ACLs restricting its access strictly to the topics it needs.

from aiokafka.helpers import create_ssl_context

producer = AIOKafkaProducer(
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-256",
    sasl_plain_username="billing-service",
    sasl_plain_password=os.environ["KAFKA_PASSWORD"],
    ssl_context=create_ssl_context(cafile="/etc/kafka/ca.pem"),
)

KRaft: Kafka without ZooKeeper

Kafka used to require a separate ZooKeeper cluster to store metadata: which topics exist, who is the partition leader, which ACLs are in effect. ZooKeeper is a separate system that has to be deployed, monitored, and fixed on its own.

As of version 3.3 (2022) there is KRaft — a native Raft-based protocol running inside the brokers themselves. ZooKeeper is no longer needed. In Kafka 4.x ZooKeeper support has been removed entirely — all new clusters run on KRaft only.

For a developer this is almost invisible: bootstrap-servers and client behavior have not changed. The only thing that changes is how the ops team deploys and maintains the cluster.

In short

AIOKafkaProducer sends, AIOKafkaConsumer with a consumer loop receives. Parallelism means several consumer tasks in the same group.
Commit offset controls when progress is recorded: auto-commit (default), after a batch, after each message.
Message headers are the place for technical metadata (trace-id, schema version). Business data goes in the payload.
The DLQ saves you from a consumer hanging on a problematic message. Retry topics are the non-blocking way, with an increasing delay.
Schema Registry stores schemas and checks compatibility on changes. BACKWARD mode — a new consumer reads old messages.
Consumer lag is the main health metric for a consumer. Growing → the consumer cannot keep up.
Producer tuning: max_batch_size, linger_ms, compression_type="zstd". Consumer tuning: max_poll_records, max_poll_interval_ms.
Security: SSL for encryption, SASL for authentication, ACL for per-topic authorization.
KRaft: as of Kafka 3.3+ ZooKeeper is not needed; in Kafka 4.x ZooKeeper support has been removed entirely.