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: компактный бинарный формат с хорошей поддержкой эволюции схем через confluent-kafka с fastavro. 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 — отдельная система, которую нужно отдельно поднимать, мониторить и чинить.

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

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

Коротко

• 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 убрана полностью.

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

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