FastAPI и Kafka: создание producer, consumer и обработка повторов

Асинхронные веб-фреймворки и распределенные очереди сообщений требуют аккуратной синхронизации жизненных циклов. При интеграции FastAPI и Apache Kafka типовые ошибки обычно связаны не с отправкой сообщений, а с фоновым потреблением: потеря смещений (offsets), дублирование полезной нагрузки при рестартах и зависание соединений при остановке контейнера.

Для построения отказоустойчивого конвейера событий на базе библиотеки aiokafka необходимо решить три инженерные задачи: привязать клиенты к менеджеру жизненного цикла FastAPI, настроить сохранение порядка событий через ключи партиционирования и реализовать надежную обработку с семантикой at-least-once.

Архитектурные требования: почему важен правильный жизненный цикл

FastAPI построен вокруг асинхронного цикла событий asyncio. Клиенты Kafka — как AIOKafkaProducer, так и AIOKafkaConsumer — удерживают постоянные сетевые сокеты к брокерам, пул буферов в памяти и внутренние фоновые корутины для heartbeat-запросов и опроса метаданных кластера.

Роль Lifespan Context Manager в FastAPI

Ранее используемые декораторы @app.on_event("startup") и @app.on_event("shutdown") признаны устаревшими. Вместо них FastAPI использует механизм lifespan на базе асинхронного контекстного менеджера (@asynccontextmanager).

Преимущества lifespan:

Логика инициализации и очистки ресурсов сосредоточена в одном блоке кода.
Гарантируется выполнение блока finally даже при возникновении необработанных исключений на этапе работы приложения.
Клиентские объекты можно безопасно сохранять в app.state, избегая глобальных переменных с неопределенным состоянием.

Асинхронный стек: почему именно aiokafka

Библиотека confluent-kafka опирается на C-библиотеку librdkafka и выполняет блокирующие вызовы, которые требуют выноса в отдельный пул потоков (run_in_executor). Неправильное управление потоками в асинхронном приложении приводит к исчерпанию пула воркеров и деградации производительности HTTP-эндпоинтов.

aiokafka полностью написана с поддержкой нативного asyncio. Это позволяет выполнять операции сетевого ввода-вывода (poll, send, commit) без блокировки основного Event Loop, в рамках которого обрабатываются HTTP-запросы сервиса.

Проектирование Producer: ключи сообщений и порядок доставки

Задача Producer — принять запрос от HTTP-слоя, сериализовать его и гарантированно передать брокеру с сохранением бизнес-контекста.

[ HTTP POST ] ──> [ FastAPI Endpoint ] ──> [ AIOKafkaProducer ]
                                                    │
                                              (Hash по ключу)
                                                    ▼
                                         [ Partition 0 | Partition 1 ]

Партиционирование по ключу для сохранения очередности

Топик Kafka состоит из одной или нескольких партиций (partitions). Брокер гарантирует строгий порядок сообщений только внутри одной конкретной партиции.

Если отправлять сообщения без ключа (key=None), продюсер распределяет их по партициям по алгоритму Round-Robin (или с помощью встроенного партиционера брокера). В этом случае два последовательных события по одной сделке (например, OrderCreated и OrderCancelled) могут попасть в разные партиции, быть прочитаны параллельно и обработаны в обратном порядке.

Передача уникального идентификатора сущности (например, order_id или user_id) в качестве ключа гарантирует:

Хеширование ключа направит все события этой сущности в одну и ту же партицию.
Консьюмер прочитает их строго в хронологическом порядке.

Защита от дублей на этапе публикации

При нестабильной сети запрос на отправку данных может успешно дойти до брокера, но подтверждение (acknowledgement) потеряется по таймауту. В результате продюсер попытается отправить пакет заново, создав дубликат в партиции.

Для исключения таких ситуаций в параметрах продюсера активируется идемпотентность:

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

При enable_idempotence=True брокер присваивает каждому продюсеру уникальный идентификатор (Producer ID, PID) и отслеживает порядковые номера (sequence numbers) сообщений, прозрачно отбрасывая повторные сетевые пакеты.

Проектирование Consumer: от at-most-once к at-least-once

Поведение консьюмера определяется тем, в какой момент фиксируется смещение (offset) прочитанного сообщения.

Смещение (Offset):
[ 0 ] ──> [ 1 ] ──> [ 2 (Committed) ] ──> [ 3 (Текущая позиция) ] ──> [ 4 ]

Механизм смещений (Offset) и committed offset

Current position (текущая позиция): смещение записи, которую консьюмер запрашивает при следующем вызове getone() или getmany().
Committed offset (зафиксированное смещение): последняя отметка, сохраненная во внутреннем системном топике __consumer_offsets. При перезапуске сервиса или ребалансировке группы консьюмер начинает вычитку с этой позиции.

Ручной коммит смещений

По умолчанию в большинстве библиотек включен автокоммит (enable_auto_commit=True), который сбрасывает смещения по таймеру (например, каждые 5 секунд). Если приложение падает в процессе выполнения сложной бизнес-логики, смещение может оказаться уже закоммиченным. В результате необработанное сообщение безвозвратно теряется — это семантика at-most-once (максимум один раз).

Для реализации семантики at-least-once (минимум один раз) автокоммит отключается:

Сообщение извлекается из топика.
Выполняется бизнес-логика (запись в базу, вызов внешнего API).
При успешном завершении вызывается явный коммит: await consumer.commit().
Если на шаге 2 произошла ошибка, коммит не вызывается. При рестарте сообщение будет получено снова.

Идемпотентность обработчика: защита от повторной доставки

Неизбежное следствие семантики at-least-once — повторная доставка (redelivery). Если консьюмер выполнил транзакцию в БД, но упал до отправки consumer.commit(), после рестарта событие придет повторно.

Чтобы избежать дублирования действий (двойное списание денег, повторная отправка уведомлений), обработчик обязан быть идемпотентным. Стандартные подходы к реализации:

Таблица дедупликации: сохранение уникального event_id в транзакционную базу данных в рамках одной транзакции с бизнес-данными (Inbox pattern).
Условные обновления (Optimistic Locking): проверка статуса сущности перед модификацией (например, переход в статус PAID разрешен только из статуса NEW).
Распределенные блокировки и кэш: быстрая проверка обработанных ключей в Redis с выставлением TTL.

Пошаговая реализация сервиса

Ниже представлен завершенный микросервис, объединяющий HTTP API для генерации событий и фоновый процесс их гарантированной обработки.

import asyncio
import json
import logging
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator

from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
from aiokafka.errors import KafkaError
from fastapi import FastAPI, HTTPException, Request, status
from pydantic import BaseModel, Field

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("event-service")

KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
TOPIC_NAME = "order-events"
CONSUMER_GROUP = "order-processing-group"

# In-memory хранилище для демонстрации идемпотентности
PROCESSED_EVENTS: set[str] = set()


class OrderEvent(BaseModel):
    event_id: str = Field(..., description="Уникальный ID события для дедупликации")
    order_id: str = Field(..., description="ID заказа, используемый как ключ партиции")
    amount: float
    status: str


class EventService:
    def __init__(self) -> None:
        self.producer: AIOKafkaProducer | None = None
        self.consumer: AIOKafkaConsumer | None = None
        self.consumer_task: asyncio.Task[None] | None = None

    async def start(self) -> None:
        self.producer = AIOKafkaProducer(
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
            enable_idempotence=True,
            acks="all",
            value_serializer=lambda v: json.dumps(v).encode("utf-8"),
            key_serializer=lambda k: k.encode("utf-8") if k else b"",
        )
        self.consumer = AIOKafkaConsumer(
            TOPIC_NAME,
            bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS,
            group_id=CONSUMER_GROUP,
            enable_auto_commit=False,
            auto_offset_reset="earliest",
            value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        )

        await self.producer.start()
        await self.consumer.start()
        logger.info("Kafka Producer и Consumer успешно запущены")

        # Запуск фонового цикла обработки
        self.consumer_task = asyncio.create_task(self.consume_loop())

    async def stop(self) -> None:
        if self.consumer_task:
            self.consumer_task.cancel()
            try:
                await self.consumer_task
            except asyncio.CancelledError:
                logger.info("Фоновая задача Consumer остановлена")

        if self.consumer:
            await self.consumer.stop()
        if self.producer:
            await self.producer.stop()
        logger.info("Соединения с Kafka корректно закрыты")

    async def consume_loop(self) -> None:
        assert self.consumer is not None
        try:
            async for record in self.consumer:
                payload = record.value
                event_id = payload.get("event_id")
                order_id = record.key.decode("utf-8") if record.key else "unknown"

                logger.info(
                    "Получено событие: key=%s, offset=%d, partition=%d",
                    order_id,
                    record.offset,
                    record.partition,
                )

                # Проверка на повторную доставку
                if event_id in PROCESSED_EVENTS:
                    logger.warning(
                        "Дубликат события %s пропущен. Фиксация offset без повторной обработки.",
                        event_id,
                    )
                    await self.consumer.commit()
                    continue

                # Имитация полезной работы
                try:
                    await self._process_business_logic(payload)
                    PROCESSED_EVENTS.add(event_id)
                    # Фиксируем смещение ТОЛЬКО после успешной обработки
                    await self.consumer.commit()
                    logger.info("Событие %s успешно зафиксировано", event_id)
                except Exception as err:
                    logger.error("Ошибка при обработке события %s: %s", event_id, err)
                    # Не делаем commit: сообщение перечитается при сбое
                    await asyncio.sleep(1)

        except asyncio.CancelledError:
            pass
        except KafkaError as exc:
            logger.error("Критическая ошибка consumer loop: %s", exc)

    async def _process_business_logic(self, payload: dict[str, Any]) -> None:
        # Имитация сетевой или дисковой задержки
        await asyncio.sleep(0.05)


event_service = EventService()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    await event_service.start()
    app.state.kafka = event_service
    try:
        yield
    finally:
        await event_service.stop()


app = FastAPI(title="Kafka Event Processing Service", lifespan=lifespan)


@app.post("/events", status_code=status.HTTP_202_ACCEPTED)
async def publish_event(event: OrderEvent, request: Request) -> dict[str, str]:
    service: EventService = request.app.state.kafka
    if not service.producer:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail="Producer не инициализирован",
        )

    try:
        # Ключ order_id гарантирует порядок в рамках одной партиции
        await service.producer.send_and_wait(
            topic=TOPIC_NAME,
            key=event.order_id,
            value=event.model_dump(),
        )
        return {"status": "accepted", "event_id": event.event_id}
    except KafkaError as exc:
        logger.error("Ошибка отправки в Kafka: %s", exc)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="Не удалось отправить событие в очередь",
        )

Тестирование отказоустойчивости и краевых сценариев

Работоспособность распределенной системы проверяется не в штатном режиме, а в условиях деградации инфраструктуры.

Сценарий 1: Рестарт Consumer во время обработки (проверка committed offset)

Цель: убедиться, что сообщение не теряется, если воркер аварийно завершает работу до момента вызова commit().

Порядок воспроизведения:

Запустите сервис и локальный брокер Kafka.

В метод _process_business_logic добавьте условие аварийного завершения:

if payload["amount"] == 999.0:
    import os, signal
    os.kill(os.getpid(), signal.SIGKILL)  # Жесткое падение процесса

Отправьте через HTTP POST событие с amount: 999.0.
Сервис немедленно упадет, не выполнив await consumer.commit().
Запустите сервис заново.

Результат: После старта консьюмер прочитает из Kafka сохраненный committed offset. Так как смещение для упавшего сообщения не было подтверждено, брокер выдаст его снова с тем же значением offset. Если обработчик идемпотентен (или ошибка исправлена), данные будут успешно сохранены без потерь.

Сценарий 2: Временная недоступность брокера Kafka

Цель: проверить поведение продюсера и консьюмера при разрыве сетевого соединения с кластером.

Порядок воспроизведения:

Запустите отправку потока событий через эндпоинт /events.
Остановите контейнер брокера командой:
```
docker stop kafka-broker
```
Отправьте очередной HTTP-запрос.

Поведение Producer: Метод send_and_wait ожидает ответа в пределах таймаута, заданного параметром request_timeout_ms (по умолчанию 40 секунд в aiokafka). Запрос заблокирует выполнение конкретного обработчика эндпоинта, после чего выбросит исключение (например, KafkaTimeoutError). Event Loop приложения при этом продолжит обрабатывать другие легкие запросы, не требующие Kafka.

Поведение Consumer: Консьюмер начнет циклически повторять попытки повторного подключения к адресам, указанным в bootstrap_servers, выводя предупреждения в лог с экспоненциальной задержкой (backoff). При восстановлении контейнера (docker start kafka-broker) консьюмер автоматически восстановит соединение и продолжит чтение с прежней позиции без необходимости перезапуска сервиса.

Чек-лист продакшен-готовности сервиса

Перед развертыванием сервиса в промышленной среде убедитесь, что соблюдены следующие требования:

Отключен автокоммит: в настройках AIOKafkaConsumer строго установлено enable_auto_commit=False.
Включена идемпотентность продюсера: параметр enable_idempotence=True предотвращает дублирование при сетевых сбоях между приложением и брокером.
Задан осмысленный ключ: события, порядок которых важен для предметной области, публикуются с ключом агрегата (order_id, client_id).
Консьюмер обернут в CancelledError: при завершении работы приложения фоновая задача получает сигнал отмены, успевая освободить ресурсы и выйти из цикла чтения.
Идемпотентность на уровне хранилища: операции записи используют уникальные индексы, конструкции INSERT ... ON CONFLICT DO NOTHING или паттерн Transactional Inbox.
Настроена обработка ядовитых сообщений (Dead Letter Queue): сообщения с некорректным форматом (которые невозможно распарсить) логируются и отправляются в специальный DLQ-топик, чтобы консьюмер не зацикливался на них вечно.

Часто задаваемые вопросы (FAQ)

Почему нельзя использовать автокоммит (enable_auto_commit=True) в критичных к данным системах?

Автокоммит фиксирует смещения каждые N миллисекунд независимо от того, завершилась ли реальная бизнес-логика обработки сообщения. Если приложение вычитало пачку сообщений и завершилось аварийно (OOM, сбой питания, падение пода) до сохранения в основную БД, Kafka посчитает эти сообщения прочитанными. Это приводит к безвозвратной потере данных (семантика at-most-once).

Что произойдет с сообщениями, если сервис упал до вызова consumer.commit()?

Сообщения останутся в партиции. Значение зафиксированного смещения (committed offset) для данной группы консьюмеров не изменится. При повторном старте сервиса (или при передаче партиции другому воркеру в процессе ребалансировки) чтение возобновится с последнего успешно закоммиченного смещения, и непрошедшие обработку события будут доставлены повторно.

Зачем задавать ключ сообщения (message key) при отправке события?

Ключ определяет, в какую именно партицию топика попадет событие (путем вычисления хеша от ключа). Kafka обеспечивает строгую очередность сообщений исключительно в рамках одной партиции. Использование общего ключа для связанных событий гарантирует, что они будут прочитаны строго по порядку одним воркером.

Как FastAPI lifespan защищает от зависших соединений при остановке контейнера?

При получении сигнала остановки (SIGTERM) сервер прекращает принимать новые HTTP-запросы и передает управление в блок после yield в контекстном менеджере lifespan. Это позволяет прервать цикл консьюмера, дождаться завершения текущей фоновой задачи и вызвать await consumer.stop() и await producer.stop(), которые корректно закрывают сокеты и отправляют брокеру уведомление о выходе из группы.

Что делать, если брокер Kafka недоступен в момент запуска FastAPI?

По умолчанию методы producer.start() и consumer.start() выбрасывают исключение при невозможности подключиться к брокеру, что приводит к аварийной остановке всего сервиса. Если API должен оставаться доступным для других операций даже без Kafka, инициализацию клиентов выносят в повторные попытки с задержкой (retry loop с backoff) либо откладывают до момента первого обращения, обрабатывая сетевые ошибки кодами 503 Service Unavailable.

Как предотвратить бесконечный цикл обработки, если в топик попало некорректное сообщение (poison pill)?

Если сообщение вызывает необработанное исключение (например, невалидный JSON), консьюмер без ручного коммита будет бесконечно падать и перечитывать его заново. Для предотвращения этой проблемы обработку оборачивают в блок try...except, ловят ошибки валидации, отправляют битое сообщение в топик ошибок (Dead Letter Queue, DLQ) и сразу после этого фиксируют смещение через await consumer.commit().

Вывод

Интеграция FastAPI и Apache Kafka требует внимательного отношения к жизненному циклу асинхронных клиентов и семантике доставки. Использование lifespan контекстного менеджера обеспечивает чистое выделение и освобождение ресурсов при старте и завершении приложения.

Отказ от автокоммита смещений в пользу ручной фиксации (await consumer.commit()) переводит систему на уровень надежности at-least-once. Однако этот подход неизбежно требует поддержки идемпотентности на уровне бизнес-логики сервиса для безопасной обработки повторных доставок при аппаратных или программных сбоях.

FastAPI и Apache Kafka: пишем надежный producer и consumer с обработкой повторной доставки