Kafka в production на Python: aiokafka, DLQ, Schema Registry, тюнинг, безопасность

← назад к разделу

Если вы уже знаете, что такое топик, партиция и группа потребителей — эта статья про следующий шаг: как подключить Kafka к Python-приложению и что нужно настроить, чтобы оно работало надёжно в production.

aiokafka: producer и consumer loop

Библиотека aiokafka — асинхронный клиент Kafka для asyncio. Для отправки сообщений используют AIOKafkaProducer, для приёма — AIOKafkaConsumer с consumer loop.

# Отправка
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),
        )

# Приём
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)
            # обработка
    finally:
        await consumer.stop()

Параллелизм — это несколько consumer-задач в одной группе: каждая создаёт свой AIOKafkaConsumer и обслуживает свой набор партиций. По умолчанию одна задача, но можно запустить больше:

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

Все три задачи живут в одном процессе и на одном ядре: это помогает, когда обработка ждёт базу или сеть, но не когда она считает. Для тяжёлой обработки запускают несколько экземпляров сервиса, по-прежнему не больше числа партиций.

Commit offset: когда Kafka «знает», что сообщение обработано

Kafka следит за тем, до какого сообщения дошёл потребитель, через offset commit. Когда этот коммит происходит — решаете вы:

• Auto-commit — каждые auto_commit_interval_ms (5 секунд, значение по умолчанию), независимо от того, обработано ли сообщение.
• После пачки — getmany() + commit() после успешной обработки всей пачки. Быстрее, но при сбое пачка перечитается целиком.
• После каждого сообщения — commit() в теле цикла. Безопаснее при сбоях, но медленнее. Нужен, когда важно закоммитить оффсет только после успешной записи в базу.

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()  # после успешной обработки всей пачки
finally:
    await consumer.stop()

Заголовки сообщений

Каждое сообщение Kafka может содержать заголовки — пары (ключ, значение) в байтах, отдельно от полезной нагрузки. Сюда кладут техническую метаинформацию, которой не место в бизнес-данных:

• X-Correlation-ID / traceparent — для распределённого трейсинга.
• X-Event-Version — версия формата события.
• X-Source-Service — откуда пришло сообщение.

# При отправке
await producer.send_and_wait(
    "orders",
    key=key,
    value=serialize(event),
    headers=[("X-Correlation-ID", correlation_id.encode())],
)

# При получении
async for record in consumer:
    headers = dict(record.headers)
    correlation_id_var.set(headers["X-Correlation-ID"].decode())  # contextvars для логов
    # обработка

Бизнес-данные (orderId, amount, status) — в payload. Всё техническое — в headers.

Dead Letter Queue: что делать с проблемными сообщениямиспросят на собеседовании

Если обработчик бросает исключение, а оффсет не коммитится, потребитель после перезапуска снова и снова получает то же самое сообщение — оффсет не двигается, потребитель встаёт. На production это означает полную остановку группы.

Решение — Dead Letter Queue (DLQ): после нескольких неудачных попыток сообщение перекладывается в отдельный топик (обычно с суффиксом .DLT), оффсет коммитится, основной потребитель продолжает работу.

Простой способ: повторы с DLQ в обработчике

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)

Здесь: 3 попытки с интервалом 1 секунда, после третьей неудачи — сообщение уходит в orders.DLT. В DLT-сообщение добавляются заголовки с текстом ошибки и именем оригинального топика.

Неблокирующий способ: retry-топики

Повторы внутри обработчика блокируют партицию: пока одно сообщение «отстаивается», остальные ждут. Альтернатива — отдельные retry-топики с нарастающей задержкой:

# orders → orders-retry-0 (1с) → orders-retry-1 (2с) → orders-retry-2 (4с) → 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)  # сообщение «отстаивается», основной топик не блокируется
        await handle_or_escalate(record)

Каждому retry-топику — свой потребитель со своей задержкой. Сообщения «отстаиваются» в retry-топиках, не блокируя основной поток.

Когда что использовать

• Временная проблема (база недоступна, внешний сервис не отвечает) → retry-топики с нарастающей задержкой.
• Постоянная проблема (невалидное сообщение, неверный формат) → без повторов, сразу в DLQ.
• Нужно различать типы ошибок → проверять тип исключения: для ValueError (невалидные данные) сразу DLQ, для ConnectionError — повторы.

Schema Registry: как не сломать соседей при изменении схемы

Когда продьюсер и потребитель — разные сервисы разных команд, любое изменение структуры события потенциально ломает потребитель. Schema Registry решает это централизованным хранилищем схем с проверкой совместимости.

Как это работает:

1. Продьюсер регистрирует схему в Schema Registry и получает числовой идентификатор (schema_id).
2. В начало каждого сообщения он пишет служебный байт и 4 байта с этим идентификатором, а дальше идёт бинарный payload без имён полей.
3. Потребитель читает schema_id, один раз скачивает схему из Schema Registry (потом кэширует) и десериализует payload.

Avro, Protobuf или JSON Schema

Стандарт в Kafka-экосистеме — Avro: компактный бинарный формат с хорошей поддержкой эволюции схем. В Python клиент Schema Registry и сериализаторы Avro живут в пакете confluent-kafka (ставится как confluent-kafka[avro], под капотом fastavro); с aiokafka их используют как обычные функции сериализации значения. Protobuf берут команды с gRPC-инфраструктурой. JSON Schema читаем глазами, но занимает в разы больше места — подходит для отладки.

Режимы совместимости

Schema Registry проверяет, не сломает ли новая версия схемы уже работающих потребителей:

• BACKWARD (по умолчанию) — новый потребитель может читать старые сообщения. Можно удалять необязательные поля и добавлять необязательные с default-значением. Нельзя добавлять обязательные поля.
• FORWARD — старый потребитель может читать новые сообщения. Зеркало BACKWARD.
• FULL — оба режима одновременно. Самый строгий.
• NONE — без проверок. Только для особых случаев.

Практическое правило: BACKWARD для топиков, где потребителей много, а продьюсер один — это типичная событийная шина.

Consumer lag: почему потребитель отстаётспросят на собеседовании

Lag — разница между последним сообщением в партиции и тем, до которого дошёл потребитель. Если lag растёт — потребитель не успевает за продьюсером. Это главная метрика здоровья потребителя в production.

Типичные причины:

1. Медленная обработка — каждое сообщение делает синхронный запрос к базе или внешнему сервису. Решение: уменьшить max_poll_records, параллелизировать обработку, перейти на пакетные запросы.
2. Мало партиций на группу — добавить consumer-задач или экземпляры сервиса (но не больше числа партиций).
3. Истекает max_poll_interval_ms — если обработка одной пачки занимает больше 5 минут (значение по умолчанию), Kafka считает потребителя мёртвым и запускает перераспределение партиций. Решение: уменьшить max_poll_records или поднять max_poll_interval_ms.
4. Блокирующий код в event loop — синхронный вызов на 30 секунд остановил все consumer-задачи, lag вырос на тысячи сообщений.

Посмотреть lag вручную:

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   ← отстаёт

Метрика для мониторинга: kafka_consumergroup_lag. Алерт: lag больше N и растёт N минут подряд.

Тюнинг производительности

Kafka работает хорошо из коробки, но несколько параметров стоит знать.

Продьюсер

Потребитель

Лучший алгоритм сжатия для большинства случаев — zstd (Kafka 2.1+): сжимает почти как gzip, работает почти как lz4.

Безопасность: SASL, SSL, ACL

По умолчанию Kafka слушает открытый порт без аутентификации — в production так не оставляют. Три слоя защиты:

• SSL/TLS — шифрование соединения. Сертификаты на брокерах и клиентах.
• SASL — аутентификация. SASL/PLAIN — логин и пароль (только поверх SSL). SASL/SCRAM-SHA-256 — безопаснее: challenge-response, пароль по сети не передаётся. SASL/OAUTHBEARER — OAuth2-токены, для интеграции с провайдером идентификации.
• ACL — авторизация: кому разрешено читать и писать в какой топик. Управляется через kafka-acls.sh.

Типичная схема: один SASL-пользователь на каждый сервис, ACL ограничивают его доступ строго до нужных топиков.

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 без ZooKeeper

Раньше Kafka требовала отдельного ZooKeeper-кластера для хранения метаданных: какие топики существуют, кто лидер партиции, какие ACL действуют. ZooKeeper — отдельная система, которую нужно отдельно поднимать, мониторить и чинить.

KRaft — собственный протокол на базе Raft внутри самих брокеров — появился в Kafka 2.8 как предварительная версия, а с 3.3 (2022) признан готовым к production. ZooKeeper больше не нужен. В Kafka 4.x поддержка ZooKeeper убрана полностью — все новые кластеры работают только на KRaft.

Для разработчика это почти незаметно: bootstrap-servers и поведение клиента не изменились. Меняется только то, как ops-команда разворачивает и обслуживает кластер.

Глубже: остановка потребителя, rebalance и confluent-kafkaрасширенноеспросят на собеседовании

consumer.stop() при ручном commit ничего не фиксирует сам: если сервис останавливают по SIGTERM посреди пачки, всё, что обработано, но не закоммичено, придёт ещё раз после рестарта. Поэтому в обработчике сигнала сначала завершают текущую пачку и вызывают commit(), и только потом stop(); при enable_auto_commit=True остановка сама отправляет последний commit.

Партиции могут отобрать и без остановки: при rebalance. Если за время обработки пачки партиция ушла другому потребителю, commit по ней упадёт с CommitFailedError, а сосед начнёт с последнего зафиксированного offset — дубли неизбежны. Чтобы зафиксировать до потери, подписываются через consumer.subscribe(["orders"], listener=...) с объектом ConsumerRebalanceListener: его on_partitions_revoked вызывается до того, как партиции отдадут, и в нём успевают дообработать и закоммитить.

aiokafka — не единственный клиент. confluent-kafka построен на librdkafka на C: он быстрее на больших потоках, именно в нём живут клиент Schema Registry и сериализаторы Avro и Protobuf, но он синхронный — в приложении на asyncio его poll() крутят в отдельном потоке или процессе. Обычный выбор: aiokafka для сервисов на FastAPI, confluent-kafka для обработчиков с большой нагрузкой на процессор, которым всё равно нужны отдельные процессы.

Коротко

• AIOKafkaProducer — отправка, AIOKafkaConsumer с consumer loop — приём. Параллелизм — несколько consumer-задач в одной группе.
• Commit offset управляет тем, когда фиксируется прогресс: auto-commit (по умолчанию), после пачки, после каждого сообщения.
• Заголовки сообщений — место для технической метаинформации (trace-id, версия схемы). Бизнес-данные — в payload.
• DLQ спасает от зависания потребителя на проблемном сообщении. Retry-топики — неблокирующий способ с нарастающей задержкой.
• Schema Registry хранит схемы и проверяет совместимость при изменениях. Режим BACKWARD — новый потребитель читает старые сообщения.
• Consumer lag — главная метрика здоровья потребителя. Растёт → потребитель не успевает.
• Тюнинг producer: max_batch_size, linger_ms, compression_type="zstd". Тюнинг consumer: max_poll_records, max_poll_interval_ms.
• Безопасность: SSL для шифрования, SASL для аутентификации, ACL для авторизации по топикам.
• KRaft: с Kafka 3.3+ ZooKeeper не нужен; в Kafka 4.x поддержка ZooKeeper убрана полностью.
• При остановке с ручным commit сначала commit(), потом stop(); перед rebalance offset фиксируют в on_partitions_revoked.

Что пощупать

Потребитель уведомлений в практикуме remodov/marketplace-system-python работает с enable_auto_commit=False и коммитит offset только после того, как уведомление и запись в processed_events закоммичены одной транзакцией; повторная доставка даёт ноль строк на ON CONFLICT DO NOTHING, и второе письмо не рождается. Payload не по контракту - ошибка обработки: offset не сдвигается, в processed_events записи нет.

Код: consumer.py, inbox.py, test_inbox.py.

Сделаем сами

Ветка step-10-events-and-contract.

Что почитать дальше

• Основы Kafka — устройство брокера, партиции, гарантии доставки, retention.
• Распределённые паттерны — Saga, Outbox, Idempotent Consumer.
• Resilience-паттерны — Circuit Breaker и Timeout в контексте событийных систем.