Практикум: остановка consumer Kafka и повторная обработка¶
Сценарии остановки из статьи.
Нужен работающий Docker. Выполните команду из
docs/blog/lab/2026-09-07-kafka-consumer-shutdown в клоне репозитория:
В зависимостях закреплены aiokafka-foundation-kit 0.1.3, aiokafka 0.14.0 и testcontainers 4.15.0
из практикума про клиентов,
а также servicewright 0.13.1. Соседняя директория должна быть на месте:
в ней находятся kafka_flow.py и функции запуска контейнера.
Драйвер создаёт и удаляет один брокер confluentinc/cp-kafka:7.6.0 в режиме KRaft.
| Файл | Назначение |
|---|---|
consumer_common.py |
Подключает общий код, задаёт минимальные настройки и контейнер servicewright |
consumer_naive.py |
Запускает цикл, который драйвер отменяет посередине пачки |
consumer_service.py |
Подключает цикл с ограниченным ожиданием к DaemonEntrypoint |
driver.py |
Создаёт десять записей и проверяет три сценария прерывания |
После второго изменения данных драйвер отменяет задачу, устанавливает событие остановки либо устанавливает событие и блокирует дальнейшую работу до истечения лимита ожидания. Синхронизация через события позволяет точно выбрать момент внутри обработки пачки.
| Сценарий | Выполненные изменения | Сохранённый offset | Первый offset нового consumer |
|---|---|---|---|
| Отменить текущую обработку | 0, 1 | Нет | 0 |
| Запросить остановку и закончить принятую пачку | 0–4 | 5 | 5 |
| Запросить остановку и исчерпать время ожидания | 0, 1 | Нет | 0 |
Проверки читают настоящий offset группы и запускают новый consumer той же группы.
Оба сценария сервиса выполняются через Service.run(..., stop=event).
TimeoutError при нехватке времени должен выйти наружу: успешное завершение
без ошибки проверку не пройдёт. Таймаут находится в run_tracking(): сам по себе
DaemonEntrypoint.drain() не ограничивает время произвольной функции воркера.
Здесь проверяются событие остановки и отмена задачи, а не SIGTERM, принудительное завершение процесса или выкладка в Kubernetes. Практикум работает и на Windows, и на Unix. Результат обработчика хранится в памяти; сохранение в БД и репликацию нужно проверять отдельно.
consumer_service.py можно запустить отдельным процессом, указав адрес Kafka в KAFKA
и при необходимости группу в GROUP (по умолчанию tracking). Топик delivery.status
нужно создать заранее. run_sync() подключит обычную обработку сигналов servicewright;
драйвер передаёт событие явно, чтобы проверки были воспроизводимыми.