Практикум: транзакционный Inbox¶
Сценарии обработки из статьи.
Нужен работающий Docker. Выполните команду из
docs/blog/lab/2026-09-07-transactional-inbox в клоне репозитория:
Скрипт импортирует event_flow.py, модели и запуск контейнеров из
практикума по доставке;
соседняя директория должна быть на месте.
requirements.txt подключает его закреплённые зависимости: omni-box 0.3.0,
aiokafka 0.14.0, SQLAlchemy 2.0.54, asyncpg 0.31.0 и testcontainers 4.15.0.
Скрипт создаёт и удаляет postgres:17-alpine и один брокер
confluentinc/cp-kafka:7.6.0 в режиме KRaft.
Обработчик записывает счёт в транзакции Inbox. Runner использует
EXACTLY_ONCE_INBOX; автоматический commit Kafka отключён,
exactly_once_commit_on_failed=False.
Проверяются следующие результаты:
- исключение после INSERT счёта откатывает и счёт, и Inbox;
- цикл воркера из статьи останавливается при неподтверждённом результате до следующей записи;
- новый consumer той же группы повторяет неудачную запись, а затем обрабатывает следующую;
- сбой при commit offset оставляет уже сохранённые счёт и готовую запись Inbox;
- при повторе обработчик пропускается, а настоящий offset Kafka подтверждается;
- другая группа имеет отдельный ключ Inbox и обрабатывает событие;
- удаление готовой строки Inbox позволяет повтору снова вызвать обработчик;
- две публикации без заголовков с постоянным ID считаются разными сообщениями по их offset.
Для сбоя offset используется подкласс consumer, который выбрасывает исключение из
commit(). Перед этим он читает PostgreSQL и проверяет, что счёт и готовая запись
Inbox уже сохранены. Исключение обработчика и явное удаление Inbox тоже заданы сценарием.
При этом Kafka, транзакции БД и перезапуск consumer настоящие.
Счета намеренно допускают повторяющиеся ID заказа, чтобы повторное создание было видно.
Каждый сценарий выводит PASS только после успешных проверок. Практикум проверяет
последовательную обработку; безопасность параллельной обработки одной партиции он не доказывает.