Kafka shutdown — aiokafka consumer stop and producer flush

Python Backend

How to correctly stop an aiokafka consumer and producer in FastAPI: manual offset commit, flushing pending messages and the right order in lifespan. Without data loss or duplicates.

Published 28 июня 2026 ~6 min read Author: Вадим Викулин

← Back to the section

Kafka is where stopping an application incorrectly hurts data the most. Kill the consumer before commit — the next run reads the same messages again. Kill the producer before flush — the last messages simply disappear. In this article we'll go through how to correctly stop aiokafka in a FastAPI application.

Why a plain kill breaks Kafka

When a process receives a stop signal (SIGTERM) and closes immediately, two things happen:

The consumer doesn't manage to commit the offset — the group coordinator doesn't know how far processing got. On the next start the consumer will start reading from the last committed offset — the same messages will arrive again.
The producer doesn't manage to send messages from its internal buffer. These messages are lost forever — the client simply doesn't know they didn't get through.

The solution: give the application time to finish its current work before exiting. That is a correct shutdown.

Consumer stop in lifespan

The typical scheme for aiokafka in FastAPI is to run the consumer as an asyncio task in lifespan, and on shutdown finish it cleanly:

from contextlib import asynccontextmanager
from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from fastapi import FastAPI
import asyncio

consumer: AIOKafkaConsumer | None = None
producer: AIOKafkaProducer | None = None
consumer_task: asyncio.Task | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global consumer, producer, consumer_task

    consumer = AIOKafkaConsumer(
        "orders.confirmed",
        bootstrap_servers="kafka:9092",
        group_id="billing-service-confirmations",
        enable_auto_commit=False,
        auto_offset_reset="earliest",
    )
    producer = AIOKafkaProducer(bootstrap_servers="kafka:9092")

    await consumer.start()
    await producer.start()

    ready = {"accepting": True}
    consumer_task = asyncio.create_task(consume_order_events(ready))

    yield

    # Shutdown — order matters
    ready["accepting"] = False
    consumer_task.cancel()
    try:
        await asyncio.wait_for(consumer_task, timeout=20.0)
    except (asyncio.CancelledError, asyncio.TimeoutError):
        pass
    await consumer.stop()    # final offset commit + leave the group
    await producer.stop()    # flush pending messages + close connection
    await engine.dispose()   # DB pool — last


app = FastAPI(lifespan=lifespan)

What await consumer.stop() does:

Stops reading (getmany() and __aiter__).
Makes a final commit of the accumulated offsets.
Sends LeaveGroup — another consumer in the group picks up the partitions without unnecessary waiting.

An important point: the task must be cancelled before calling consumer.stop(). Otherwise the task keeps reading messages while stop is already running — this creates a race.

Why auto commit can't be left enabled

enable_auto_commit=True is the default setting in many clients. It's convenient but dangerous: aiokafka commits the offset on a timer, without waiting for message processing to finish.

The loss scenario:

A message arrived, its long handle() started.
The timer fired, the offset was committed.
SIGTERM — the process died in the middle of handle().
On the next start the message won't arrive — the offset is already committed. The data is lost.

The correct approach — enable_auto_commit=False and an explicit await consumer.commit() after each message is successfully processed:

async def consume_order_events(ready: dict) -> None:
    async for msg in consumer:
        try:
            await handle_order_confirmed(msg)
            await consumer.commit()
        except asyncio.CancelledError:
            # On shutdown: don't commit the offset, the message will arrive again
            raise
        except Exception:
            logger.exception(
                "order_event_processing_failed",
                topic=msg.topic,
                partition=msg.partition,
                offset=msg.offset,
            )
            # Don't commit — on the next start processing will repeat

asyncio.CancelledError on shutdown needs to be re-raised (raise), not suppressed. If you suppress it, the task will keep working after cancel, and the shutdown will hang.

Batch processing

If there are many messages, it's convenient to read them in batches via getmany(). The ready["accepting"] flag lets you stop reading new batches when a stop signal is received:

async def consume_order_events(ready: dict) -> None:
    while ready["accepting"]:
        batch = await consumer.getmany(timeout_ms=1000, max_records=50)
        for tp, messages in batch.items():
            for msg in messages:
                await handle_order_confirmed(msg)
        if batch:
            await consumer.commit()

On shutdown lifespan sets ready["accepting"] = False — the current batch is finished processing, a new one isn't started. The task exits the loop cleanly. If a CancelledError arrives inside the loop — the offset isn't committed, and the whole batch will arrive again on the next start.

Long calls in the handler — a common mistake

Inside the consumer loop you shouldn't make long HTTP requests with retries:

# Dangerous: if this is interrupted halfway — we get a duplicate
async def handle_order_confirmed(msg):
    event = OrderConfirmedEvent.model_validate_json(msg.value)
    await payment_client.charge(event.order_id, event.total)    # up to 35 seconds with retries
    await notification_client.send(event.order_id)              # another 15 seconds

The problem: if SIGTERM arrives while payment_client.charge is in progress — the payment may have gone to the bank, but the offset isn't committed yet. The next start repeats the processing — a second payment.

The correct way — do only a local transaction and write a job into the outbox. A separate worker will then send the HTTP requests with the necessary retries:

async def handle_order_confirmed(msg):
    event = OrderConfirmedEvent.model_validate_json(msg.value)

    async with session_factory() as session:
        async with session.begin():
            exists = await session.scalar(
                select(ProcessedEvent.id)
                .where(ProcessedEvent.event_id == event.event_id)
                .where(ProcessedEvent.consumer_group == "billing-confirmations")
            )
            if exists:
                return  # already processed, skip

            session.add(ProcessedEvent(
                event_id=event.event_id,
                consumer_group="billing-confirmations",
            ))
            session.add(OutboxEvent(
                aggregate_id=str(event.order_id),
                event_type="ChargePaymentRequested",
                payload={"order_id": str(event.order_id), "total": str(event.total)},
            ))

Such a handler works in under 100 milliseconds — shutdown is safe at any moment.

Producer flush

The producer also needs to be stopped explicitly. await producer.stop() does two things: first flush() — finishes sending all messages from the internal buffer to the broker, then closes the connection.

producer = AIOKafkaProducer(
    bootstrap_servers="kafka:9092",
    acks="all",
    enable_idempotence=True,
)

On shutdown in lifespan:

await producer.stop()  # flush all pending messages + close connection

Without this call, messages that the client accepted but hasn't yet sent will be lost.

For critical events send_and_wait is convenient — it waits for confirmation from the broker right at send time:

async def publish_product_price_updated(product_id: str, new_price: Decimal):
    event = ProductPriceUpdatedEvent(
        product_id=product_id,
        new_price=str(new_price),
    )
    await producer.send_and_wait(
        "products.price-updated",
        value=event.model_dump_json().encode(),
        key=product_id.encode(),
    )

For high load — use send() (doesn't wait for ack) plus an explicit await producer.flush() at the end of the batch, or rely on producer.stop() at shutdown.

Shutdown order

The sequence in the lifespan shutdown matters:

Set the flag ready["accepting"] = False.
Cancel the consumer task: consumer_task.cancel().
Wait for the task to finish: await asyncio.wait_for(consumer_task, timeout=20.0).
await consumer.stop() — final offset commit, leave the group.
await producer.stop() — flush pending messages, close the connection.
await engine.dispose() — close the database pool.

If you change the order — for example, close the database pool before consumer.stop() — a handler trying to write an outbox event to the database on shutdown will get a connection error.

In short

enable_auto_commit=False is mandatory — with auto commit the offset is committed before processing finishes, and data is lost on failure.
An explicit await consumer.commit() after each successfully processed message or batch.
asyncio.CancelledError must be re-raised, not suppressed — otherwise the task won't finish on cancel.
Long HTTP calls with retries shouldn't be inside the consumer loop — only a local transaction plus outbox.
await producer.stop() flushes all pending messages before closing the connection.
The order in the lifespan shutdown: consumer first (commit + LeaveGroup), then producer (flush), the DB pool last.