Kafka в Python: отправка, обработка и остановка сервиса¶

Сервис доставки публикует статусы посылок в Kafka, а сервис отслеживания читает их и показывает покупателю, где сейчас заказ. Самое хрупкое место здесь не отправка, а момент, когда consumer сообщает Kafka, что запись обработана.

Если consumer фиксирует offset раньше, чем обработчик закончил работу, после перезапуска эти записи будут пропущены. Consumer, остановленный посреди пачки, должен либо доделать её, либо не фиксировать вовсе, и тогда её заново прочитает следующий экземпляр.

Соберём оба сервиса на aiokafka-foundation-kit, уроним обработчик, остановим consumer посреди пачки и каждый раз проверим, откуда продолжит следующий экземпляр.

Отправляем статус доставки¶

Сначала договоримся о формате сообщения:

aiokafka-foundation-kit даёт настройки, сериализацию JSON и управление жизненным циклом клиентов; модели настроек на Pydantic ставятся с extra [models]. Код взят из kafka_flow.py практикума с клиентами. Сервис доставки открывает один producer на всё время работы:

from contextlib import asynccontextmanager

from aiokafka_foundation_kit import producer_lifecycle
from aiokafka_foundation_kit.contrib.models import BaseKafkaProducerSettings

TOPIC = "delivery.status"


@asynccontextmanager
async def delivery_producer(bootstrap):
    settings = BaseKafkaProducerSettings(
        bootstrap_servers=bootstrap,
        acks="all", enable_idempotence=True, compression_type="gzip",
    )
    async with producer_lifecycle(settings) as producer:
        yield producer

acks="all", идемпотентность и gzip библиотека включает по умолчанию, здесь они записаны явно. acks="all" ждёт подтверждения от всех синхронных реплик (ISR), так что при min.insync.replicas=1 хватает одного лидера.

async def publish_status(producer, delivery_id, status, revision):
    return await producer.send_and_wait(
        TOPIC,
        value={"delivery_id": delivery_id, "status": status, "revision": revision},
        key=delivery_id.encode("utf-8"),
    )

send_and_wait() возвращает подтверждение брокера с партицией и offset; это не значит, что сервис отслеживания уже применил обновление. Идемпотентность закрывает только внутренние повторы producer (гарантии producer в aiokafka): если ваш код сам повторит publish_status() после таймаута, обновление может попасть в топик дважды. Чтобы сохранить данные вместе с сообщением, нужен Outbox.

Обновляем витрину отслеживания, потом делаем commit¶

Обработчик хранит последний статус каждой доставки и пропускает повторные и устаревшие версии:

class TrackingView:
    def __init__(self):
        self.deliveries = {}

    async def process(self, message):
        event = message.value
        delivery_id = event["delivery_id"]
        previous = self.deliveries.get(delivery_id)
        if previous is None or event["revision"] > previous["revision"]:
            self.deliveries[delivery_id] = event

Словарь нужен практикуму, чтобы было что проверить; настоящей витрине нужна надёжная база данных. Правило «оставить наибольший revision» подходит для полных снимков статуса. Если важен каждый переход (например, на каждый отправляется уведомление), наибольшего revision мало: событиям нужен номер без пропусков, а обработчик должен их замечать. Consumer делает commit вручную:

from aiokafka_foundation_kit import consumer_lifecycle
from aiokafka_foundation_kit.contrib.models import BaseKafkaConsumerSettings


def tracking_settings(bootstrap, group="tracking"):
    return BaseKafkaConsumerSettings(
        bootstrap_servers=bootstrap, group_id=group,
        enable_auto_commit=False, auto_offset_reset="earliest",
        max_poll_records=5, max_poll_interval_ms=30_000,
    )

earliest срабатывает, только если у группы нет сохранённого offset; новый экземпляр продолжает с сохранённой позиции. Цикл получает уже запущенный consumer из consumer_lifecycle; как его открыть, показано в разделе про остановку:

import asyncio


async def consume_batches(consumer, process, stop):
    while not stop.is_set():
        batches = await consumer.getmany(timeout_ms=500, max_records=5)
        async with asyncio.timeout(10):
            for partition, messages in batches.items():
                for message in messages:
                    await process(message)
                if messages:
                    await consumer.commit({partition: messages[-1].offset + 1})

Каждый опрос возвращает не больше пяти записей, а на обработку и commit пачки отведено десять секунд. Offset партиции фиксируется, когда все её записи обработаны: после записей 0–4 сохраняется offset 5, то есть следующий к чтению. Если обработка упала или не уложилась в лимит, цикл завершается, и offset за упавшей записью не фиксируется; партиции, обработанные раньше, прогресс сохраняют, а повторы отсекает проверка revision. Практикум проверяет одну партицию:

Последняя строка показывает, почему auto-commit не подходит: он сдвигает точку восстановления за ещё не сделанную работу, и обычный перезапуск эти записи пропустит. После сбоя закройте consumer или верните его к сохранённому offset, прежде чем читать дальше: позиция в памяти уже за неудачной пачкой. Запись, которая падает всегда, будет останавливать партицию при каждом перезапуске, так что заранее решите, пропускать её или отправлять в dead-letter. Хуже всего с битым JSON: он падает внутри getmany(), обработчик его не видит, а consumer_lifecycle() в 0.1.3 не принимает свой десериализатор (#50). Пока это не исправлено, создавайте consumer через create_async_kafka_consumer(settings, topics, value_deserializer=...) с десериализатором, который пропускает нераспознанные байты как есть, и отправляйте их в dead-letter из обработчика. Неудачный commit тоже должен останавливать цикл, а ребалансировка может забрать партицию, пока работа ещё идёт (consumer group в aiokafka).

Создаём топик до запуска сервиса¶

Число партиций потом почти не изменить: их можно только добавить, и тогда ключи переедут в другие партиции, а порядок по ключу нарушится. Партиции, срок хранения и права задаёт команда-владелец, а не сервис, который стартовал первым; а сервису, который создаёт топики, к тому же нужны права на создание. Пусть delivery.status создаёт шаг развёртывания команды доставки:

from aiokafka_foundation_kit import TopicConfig, ensure_topics_async


async def provision_delivery_topic(bootstrap, *, partitions=3, replicas=1):
    settings = BaseKafkaProducerSettings(bootstrap_servers=bootstrap)
    await ensure_topics_async([
        TopicConfig(
            name=TOPIC, num_partitions=partitions, replication_factor=replicas,
            topic_configs={"retention.ms": "604800000"},
        ),
    ], settings)

Одна реплика годится только для единственного брокера практикума. ensure_topics_async() создаёт отсутствующие топики и не приводит существующие к нужному виду, это видно в практикуме с топиками:

Создание списка не атомарно: для [valid_topic, invalid_topic, next_topic] первый остаётся созданным, второй падает, а до третьего дело не доходит. Поэтому шаг развёртывания должен ещё проверять число партиций и настройки существующего топика и сообщать о частичном результате до запуска воркеров. С producer_lifecycle(..., topics=[...]) топики создаются, только если передать ещё и auto_create_topics=True. Отдельно: на брокере в практикумах выключено auto.create.topics.enable, поэтому забытый шаг создания топика проявится ошибкой, а не тихо созданным топиком с настройками по умолчанию.

Останавливаемся посреди пачки¶

Во время выкатки старый экземпляр сервиса отслеживания прочитал пять записей и обработал две, когда сработало событие остановки. Если доделать оставшиеся три, он сохранит offset 5; если отменить работу, пачка останется для повторной обработки:

async def run_tracking(bootstrap, process, stop, *, group="tracking", grace=5):
    async with consumer_lifecycle(
        tracking_settings(bootstrap, group), topics=(TOPIC,),
    ) as consumer:
        worker = asyncio.create_task(consume_batches(consumer, process, stop))
        stopping = asyncio.create_task(stop.wait())
        try:
            await asyncio.wait({worker, stopping}, return_when=asyncio.FIRST_COMPLETED)
            async with asyncio.timeout(grace):
                await worker
        finally:
            worker.cancel()
            stopping.cancel()
            await asyncio.gather(worker, stopping, return_exceptions=True)

consume_batches() проверяет stop перед следующим опросом, поэтому если остановку запросили во время обработки, цикл доделает текущую пачку и новых записей не возьмёт. Когда остановка запрошена, grace даёт этой пачке пять секунд, это меньше десятисекундного лимита на пачку, поэтому первым срабатывает grace; тогда работа отменяется, offset недоделанной партиции не фиксируется, и consumer закрывается. В сервисе на servicewright из практикума DaemonEntrypoint передаёт событие остановки в run_tracking(), а servicewright переходит к своей фазе остановки, только когда функция вернёт управление, поэтому drain_grace_seconds это ожидание не ограничивает (servicewright #88): ограничивает только grace. Когда grace истекает, run_tracking() выбрасывает TimeoutError, и сервис завершается с ошибкой. При десяти записях в одной партиции и grace 0,05 секунды для случая с истечением времени:

---
config:
  theme: default
  look: classic
  flowchart:
    useMaxWidth: false
    wrappingWidth: 150
    padding: 12
    nodeSpacing: 24
    rankSpacing: 32
---
flowchart TD
    accTitle: Сначала закончить принятую пачку, потом остановиться
    accDescr: После обработки пачки фиксируется следующий offset. Сбой или истёкший grace оставляют пачку без commit для повторной обработки, и consumer закрывается.
    A["Прочитать до 5 записей"]
    B["Применить статусы доставки"]
    C["Сохранить следующий offset"]
    D["Остановка запрошена?"]
    E["Закрыть consumer"]
    F["Оставить пачку без commit"]
    A --> B --> C --> D
    D -->|"нет"| A
    D -->|"да"| E
    B -->|"сбой или grace истёк"| F --> E

Проверки построены на событии остановки, настоящем брокере и новых экземплярах consumer; выкатку в Kubernetes они не воспроизводят. Порядок остановки при развёртывании разобран в статье о жизненном цикле сервиса.

Фиксируем только сделанную работу¶

Фиксируйте только завершённую работу, а при сбое выходите из цикла. Следите за отставанием по сохранённому offset (конец лога минус сохранённый offset), а не только по позиции чтения: в практикуме consumer, который прочитал пять записей при сохранённом offset 0, показывает отставание позиции 0 и отставание по commit 5. aiokafka-foundation-kit берёт на себя настройки, JSON-клиентов и жизненный цикл; контракт топика, надёжное сохранение результата, решение о commit и политика для записей, которые всегда падают, остаются в вашем приложении.

Примеры и практикумы¶

Проверено на Python 3.13, aiokafka-foundation-kit 0.1.3, aiokafka 0.14.0 и servicewright 0.13.1 с confluentinc/cp-kafka:7.6.0 в режиме KRaft.

• Практикум: JSON, commit и повторная доставка
• Практикум: создание топиков и одновременный запуск
• Практикум: consumer доделывает пачку и не укладывается в бюджет остановки