Skip to content

Consuming Kafka Topics with aiokafka

Kafka's consumer model is unusual enough that transplanting habits from a task queue produces a consumer that looks right and loses messages. There is no per-message acknowledgement: progress is a single offset per partition, so committing means "everything before this point is done". Parallelism is bounded by the partition count, not by how many consumers you start. And ordering exists only within a partition, which makes the message key — not the consumer code — the thing that decides what is processed in order. This guide builds a consumer against a live Kafka 4.1 broker with aiokafka 0.14, verifying the redelivery and assignment behaviour rather than describing it.

Prerequisites

  • Python 3.11+ with aiokafka (pip install aiokafka) and a reachable broker.
  • Queue fundamentals from Async Queue Management — a Kafka partition is a durable, replayable version of the same idea.
  • Bounded processing from Worker Pool Implementations, because a consumer that fetches faster than it processes needs a limit.
One message, end to end 5 stages from producer.send to commit the offset. One message, end to end producer.send key picks the partition partition log append only, ordered group member owns whole partitions handler runs per message or batch commit the offset progress is recorded Order exists within a partition only, which is why the key matters more than anything else.

1. Produce with keys, acks and idempotence

The producer's defaults are not the ones a service wants. Three settings matter:

producer = AIOKafkaProducer(
    bootstrap_servers="kafka:9092",
    acks="all",                                        # wait for the in-sync replicas
    enable_idempotence=True,                           # no duplicates from internal retries
    linger_ms=5,                                       # batch briefly: throughput for 5 ms
)
await producer.start()
try:
    await producer.send_and_wait("orders", value, key=f"customer-{customer_id}".encode())
finally:
    await producer.stop()

acks="all" is the difference between "the leader has it" and "the replicas have it"; enable_idempotence=True stops the producer's own retries from duplicating messages; linger_ms trades a few milliseconds of latency for batching. Sending 100 messages this way took 0.02 s.

The key decides the partition, and therefore the ordering guarantee. Those 100 messages, keyed by five customer ids, landed on two of the three partitions — a reminder that hashing spreads keys unevenly, and that a small number of keys will not use all your partitions. Everything for one key stays in one partition, and therefore stays in order.

stop() matters: it flushes buffered messages. A producer garbage-collected without it silently drops whatever linger_ms was still holding.

Verify: send_and_wait returns metadata whose partition is stable for a given key.

2. Fetch batches and commit deliberately

Turn auto-commit off. Fetch with getmany, which returns a dict keyed by partition and lets you process each partition's messages in order:

consumer = AIOKafkaConsumer(
    "orders", bootstrap_servers="kafka:9092", group_id="order-processor",
    enable_auto_commit=False,                          # commit when the work is done
    auto_offset_reset="earliest",                      # a new group starts from the beginning
    max_poll_records=50,
)
await consumer.start()
try:
    while True:
        batches = await consumer.getmany(timeout_ms=1000, max_records=50)
        for tp, messages in batches.items():
            for message in messages:
                await handle(message)
            await consumer.commit({tp: messages[-1].offset + 1})   # +1: the NEXT offset
finally:
    await consumer.stop()

Consuming 100 messages this way took 0.08 s. The + 1 is the convention every Kafka client shares: a committed offset is the next message to read, not the last one read. Committing messages[-1].offset replays the final message of every batch forever.

async for message in consumer is the simpler alternative and is fine when messages are independent and cheap; getmany is better when you want to batch database writes, because one commit per batch is one round trip instead of fifty.

Verify: await consumer.committed(tp) returns one more than the last processed offset.

The shape of a reliable consume loop 5 ordered steps. The shape of a reliable consume loop getmany(timeout_ms, max_records) a dict keyed by partition process the batch in order, per partition commit last offset + 1 the next offset to read handle failures retry, or route to a dead letter stop() on shutdown leaves the group cleanly Committing offset + 1 is the convention: an offset means "the next message to read".

3. Confirm the redelivery semantics yourself

The reason to commit after processing is at-least-once delivery, and it is worth proving on your own broker:

first run consumed 10 without committing: ['m0', 'm1', 'm2']...
after restart, redelivered: ['m0', 'm1', 'm2']... overlap=10

A consumer that read ten messages and stopped without committing saw all ten again on restart. That is the guarantee — and its consequence: your handler must be idempotent, because a crash between processing and committing replays the batch. Deduplicate by a natural key, or make the write an upsert; classifying retryable errors covers the same reasoning for HTTP.

The opposite arrangement — committing before processing — converts redelivery into silent loss, which is worse in every system that has ever been debugged at 3 a.m.

Verify: kill the consumer mid-batch and confirm the uncommitted messages are reprocessed.

What each commit strategy actually promises A grid of 4 rows by 2 columns. What each commit strategy actually promises strategy on a crash cost enable_auto_commit=True unprocessed messages lost none: fastest commit after the batch the batch is redelivered one request per batch commit after each message one message redelivered a request per message commit before processing silent message loss never do this Verified: 10 messages consumed without committing were all redelivered after a restart.

4. Understand what adding consumers does

Partitions are the unit of parallelism. Two members of the same group, against a three-partition topic, received:

partition assignment across two members: {'c1': [1], 'c2': [0, 2]}

Three partitions, two members, split 1 and 2 — never finer. A third member would take one partition each; a fourth would sit idle, because there is no partition left to assign. Choosing the partition count is therefore a capacity decision made in advance: raising it later redistributes keys across partitions and breaks the per-key ordering for messages already in flight.

Every join and leave triggers a rebalance, during which consumption pauses. Two practical consequences: do not use a random group_id per process restart, and call stop() on shutdown so the member leaves cleanly rather than waiting out the session timeout.

Verify: start more consumers than partitions and confirm the extras are assigned nothing.

What adding consumers does 2 columns contrasting members <= partitions, members > partitions. What adding consumers does members <= partitions throughput scales each member owns whole partitions verified: 3 partitions split 2 and 1 rebalance on join and leave ordering preserved per key members > partitions the extras idle no partition left to assign added capacity does nothing still costs a rebalance repartitioning is the only fix Choose the partition count for the parallelism you will need; raising it later reshuffles keys.

5. Measure lag, and keep the handler off the fetch path

Consumer lag — how far behind the end of each partition you are — is the health metric for a Kafka consumer:

end = await consumer.end_offsets(list(consumer.assignment()))
lag = {tp: end[tp] - await consumer.position(tp) for tp in consumer.assignment()}

Measured on a partially consumed topic: {partition 0: 20}. Export it per partition, not just as a total — one stuck partition is invisible in an aggregate, and a stuck partition usually means one key's messages are failing repeatedly.

Slow handlers create a second problem. Kafka expects the consumer to poll regularly; a handler that blocks the loop long enough triggers a rebalance, and the partitions you were working on are reassigned mid-batch. Keep processing off the fetch path when handlers are slow:

queue: asyncio.Queue = asyncio.Queue(maxsize=1000)     # bounded: back-pressure
# fetch loop puts batches on the queue; worker tasks drain it and report completion

Only commit an offset once every message before it has completed, which for a worker pool means tracking completion per partition rather than committing whatever finished last. That bookkeeping is the price of parallel processing inside a partition — and the reason many services keep one task per partition and scale by partitioning instead.

Verify: lag is exported per partition and returns to zero after a burst.

Verification

A Kafka consumer is production-ready when:

  • Auto-commit is off and offsets are committed after processing, as offset + 1.
  • Handlers are idempotent, verified by replaying a batch deliberately.
  • Group membership is stable: a fixed group_id, and stop() on shutdown.
  • Partition count matches intended parallelism, with no idle members.
  • Lag is exported per partition and alerted on.
  • Slow handlers do not block fetching, with bounded hand-off to workers.

Pitfalls & edge cases

  • Committing offset instead of offset + 1. The last message of every batch is reprocessed forever.
  • auto_offset_reset="latest" on a new group. The consumer silently skips everything already in the topic.
  • A random group_id. Every restart re-reads the whole topic and rebalances the real group.
  • Blocking the loop in a handler. Missed polls cause a rebalance and duplicate processing.
  • Unbounded internal queues. Kafka's fetch is fast; without a bound, a slow handler turns a backlog into memory exhaustion.
  • Assuming global ordering. Order holds within a partition only; if two messages must be ordered, they must share a key.

Frequently Asked Questions

How do I consume a Kafka topic in asyncio?

Create an AIOKafkaConsumer with a group_id and enable_auto_commit=False, start it, and loop on getmany(timeout_ms=..., max_records=...), which returns messages grouped by partition. Process each partition's messages in order, then commit the last offset plus one for that partition.

Why do I get duplicate messages after restarting a Kafka consumer?

Because the offsets for those messages were never committed. Kafka gives at-least-once delivery when you commit after processing: verified here, ten messages consumed without a commit were all redelivered on restart. Make handlers idempotent rather than trying to eliminate the duplicates.

Should I commit after every message or after every batch?

After every batch in most services — one commit request instead of one per message, with at most one batch replayed after a crash. Per-message commits are worth it only when reprocessing a message is expensive or externally visible.

Does adding more consumers increase Kafka throughput?

Only up to the partition count. Partitions are assigned whole, so with three partitions a fourth group member is assigned nothing and idles. Verified with two members on three partitions: one got a single partition, the other got two.

How do I measure Kafka consumer lag in Python?

Compare the end offsets with your current positions: end = await consumer.end_offsets(list(consumer.assignment())), then end[tp] - await consumer.position(tp) for each partition. Export it per partition, because one stuck partition disappears in an aggregate.