Implementing a Kafka Consumer in FastAPI — aiokafka AIOKafkaConsumer Guide

This post is a follow-up to Using Kafka in FastAPI.
Please refer to that article first for Producer implementation and basic setup.

In the previous post, we implemented a Kafka Producer in a FastAPI service.
In this post, we’ll implement a Consumer using AIOKafkaConsumer.
We’ll subscribe to messages, manually commit offsets, and handle errors safely.

1. Consumer vs Producer Lifecycle

A Producer is called per request, while a Consumer runs continuously in the background as soon as the application starts.

Producer: request → publish → response
Consumer: app start → subscribe → loop(poll → process → commit) → app stop

In FastAPI, we start the Consumer as an asyncio task inside the lifespan, and shut it down gracefully when the app stops.

2. KafkaConsumer Wrapper Implementation

Wrap AIOKafkaConsumer to handle lifecycle management and offset commits.

# src/my_service/services/kafka_consumer.py
import asyncio
import logging
from collections.abc import Callable, Awaitable
from aiokafka import AIOKafkaConsumer
from aiokafka.errors import KafkaError
from ..core.config import KafkaSettings

logger = logging.getLogger(__name__)

MessageHandler = Callable[[bytes, bytes | None], Awaitable[None]]


class KafkaConsumer:
    def __init__(
        self,
        settings: KafkaSettings,
        topics: list[str],
        group_id: str,
        handler: MessageHandler,
    ) -> None:
        kwargs: dict = {
            "bootstrap_servers": settings.bootstrap_servers,
            "security_protocol": settings.security_protocol,
            "group_id": group_id,
            "auto_offset_reset": "earliest",
            "enable_auto_commit": False,  # Manual commit required
        }
        if settings.security_protocol == "SASL_SSL":
            kwargs["sasl_mechanism"] = settings.sasl_mechanism
            kwargs["sasl_plain_username"] = settings.sasl_username
            kwargs["sasl_plain_password"] = settings.sasl_password

        self._consumer = AIOKafkaConsumer(*topics, **kwargs)
        self._handler = handler
        self._running = False
        self._task: asyncio.Task | None = None

    async def start(self) -> None:
        await self._consumer.start()
        self._running = True
        self._task = asyncio.create_task(self._consume_loop())
        logger.info("Kafka consumer started")

    async def stop(self) -> None:
        self._running = False
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
        await self._consumer.stop()
        logger.info("Kafka consumer stopped")

    async def _consume_loop(self) -> None:
        while self._running:
            try:
                async for message in self._consumer:
                    if not self._running:
                        break
                    try:
                        await self._handler(message.value, message.key)
                        # Commit only on success
                        await self._consumer.commit()
                    except Exception as e:
                        logger.error(
                            "Message processing failed, skipping commit",
                            extra={
                                "topic": message.topic,
                                "partition": message.partition,
                                "offset": message.offset,
                                "error": str(e),
                            },
                        )
                        # If not committed, message will be reprocessed on restart
            except KafkaError as e:
                logger.error("Kafka error in consume loop", extra={"error": str(e)})
                await asyncio.sleep(5)  # wait and retry

Key Design Principles

• enable_auto_commit=False — commit manually after processing
• Do not commit on failure → message will be retried after restart
• On KafkaError, keep the loop alive and retry connection

3. Event Schema and Handler

The handler deserializes the message and delegates to business logic.

# src/my_service/services/kafka_event_handler.py
import json
import logging
from ..schemas.events import MyItemEvent
from .item_service import process_item_event

logger = logging.getLogger(__name__)


async def handle_item_event(value: bytes, key: bytes | None) -> None:
    """Deserialize Kafka message and delegate to business logic."""
    try:
        payload = json.loads(value.decode("utf-8"))
        event = MyItemEvent.model_validate(payload)
    except Exception as e:
        logger.error(
            "Failed to deserialize message, dropping",
            extra={"error": str(e), "raw": value[:200]},
        )
        # Deserialization failure → commit and skip (retry won’t help)
        return

    logger.info(
        "Processing event",
        extra={"event_type": event.event_type, "item_id": event.item_id},
    )
    await process_item_event(event)
    logger.info("Event processed", extra={"item_id": event.item_id})

Handling Deserialization Failures

Invalid messages will continue to fail even if retried.
In this case, log the error and skip by committing the offset.
If needed, you can also send them to a separate DLT for later analysis.

4. Topic and Group Constants

# src/my_service/core/messaging_constants.py
class MessagingConstants:
    TOPIC_MY_ITEM_CREATED = "myapp.my.item.created"
    TOPIC_MY_ITEM_DELETED = "myapp.my.item.deleted"

    GROUP_MY_SERVICE = "my-service-cg"

5. Managing Consumer as a Singleton

# src/my_service/core/dependencies.py
from ..services.kafka_consumer import KafkaConsumer
from ..services.kafka_event_handler import handle_item_event
from ..core.messaging_constants import MessagingConstants
from .config import KafkaSettings

_consumer: KafkaConsumer | None = None


async def init_kafka_consumer(settings: KafkaSettings) -> None:
    global _consumer
    _consumer = KafkaConsumer(
        settings=settings,
        topics=[
            MessagingConstants.TOPIC_MY_ITEM_CREATED,
            MessagingConstants.TOPIC_MY_ITEM_DELETED,
        ],
        group_id=MessagingConstants.GROUP_MY_SERVICE,
        handler=handle_item_event,
    )
    await _consumer.start()


async def shutdown_kafka_consumer() -> None:
    global _consumer
    if _consumer:
        await _consumer.stop()
    _consumer = None

---

6. Register Consumer in App Lifespan

Manage both Producer and Consumer within the application lifespan.

# src/my_service/core/app.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .config import KafkaSettings
from .dependencies import (
    init_kafka_publisher, shutdown_kafka_publisher,
    init_kafka_consumer, shutdown_kafka_consumer,
)


@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = KafkaSettings()
    await init_kafka_publisher(settings)
    await init_kafka_consumer(settings)

    yield

    await shutdown_kafka_consumer()
    await shutdown_kafka_publisher()


app = FastAPI(lifespan=lifespan)

Stop the Consumer first, then the Producer.
If the Producer is closed first while the Consumer is still processing messages, it may cause issues.

7. Prevent Duplicate Processing with idempotencyKey

In an at-least-once delivery model, the same message may be processed more than once.
Use event_id (UUID) to prevent duplicate processing.

# src/my_service/services/item_service.py
async def process_item_event(event: MyItemEvent) -> None:
    async with db_session() as session:
        # Check if already processed
        if await processed_event_repo.exists(session, event.event_id):
            logger.info("Duplicate event, skipping", extra={"event_id": event.event_id})
            return

        # Execute business logic
        await item_repo.upsert(session, event)

        # Record processed event
        await processed_event_repo.create(session, event.event_id)
        await session.commit()

8. Checklist for Adding Consumer to a New Service

• [ ] Implement KafkaConsumer wrapper (AIOKafkaConsumer, enable_auto_commit=False)
• [ ] Implement event handler (deserialize → delegate to business logic)
• [ ] Add topic/group constants in MessagingConstants
• [ ] Add singleton consumer management in dependencies.py
• [ ] Register init_kafka_consumer() / shutdown_kafka_consumer() in app.py
• [ ] Define strategy for deserialization failures (skip or send to DLT)
• [ ] Prevent duplicates using event_id (at-least-once model)

Related Posts

• Kafka Core Concepts
• Kafka Infrastructure Setup & Topic Design
• Using Kafka in FastAPI — Producer implementation
• Kafka Testing Strategy — EmbeddedKafka, testcontainers