Доставка, транзакции и exactly once

Гарантия доставки описывает, что происходит с результатом при повторе или отказе. Если consumer сохранил результат, но не успел выполнить commit позиции, восстановление может повторить обработку. Например, операция counter = counter + 1 увеличит счётчик ещё раз. Приёмник должен согласовать такой эффект с повторной доставкой.

Гарантию нужно привязать к результату

At most once допускает пропуск обработки, но избегает повторной попытки в выбранной границе. At least once допускает повтор, чтобы событие могло быть обработано после восстановления. Exactly once, EOS, означает один наблюдаемый эффект каждого события в явно заданном контуре.

Эти формулировки относятся к принятому набору отказов и правилам хранения. Если история уже удалена или приложение окончательно отбросило событие, требовать будущую обработку бессмысленно. Поэтому вместе с гарантией называют вход, результат, место checkpoint и способ восстановления.

Входом может быть топик Kafka, а результатом — строки внешней таблицы или изменение счётчика. «Producer не продублировал запись» и «consumer не повторил эффект» обозначают разные гарантии. Транспортная идемпотентность защищает журнал от retry producer; итог consumer требует отдельного решения.

Между результатом и commit есть окно отказа

Пусть consumer читает запись 4, а сохранённая следующая позиция пока равна 4.

Первый порядок показывает риск at most once, второй — at least once. Если второй вариант имеет устойчивый retry и история ещё доступна, повтор помогает восстановить обработку. Но приёмник должен сделать повтор безопасным.

На схеме приёмник защищён уникальным event_id: повторная попытка вставки оставляет одну строку. Без этой защиты повтор доставки способен повторить и эффект.

Автоматический commit позиции не анализирует успешность твоего SQL. Если consumer раздаёт события фоновым задачам и продолжает poll, позиция может продвинуться раньше записи результата. Ручной commit после завершения обработки даёт контроль, но сохраняет окно между двумя независимыми системами. Назвать ручной commit «exactly once» поэтому нельзя.

Kafka transactions связывает выход и offsets

Для контура Kafka → обработка → Kafka транзакционный producer может атомарно опубликовать выходные записи и сохранить входные offsets группы. Тогда либо видны результат и новая позиция вместе, либо не виден ни один из них.

Получить входные записи
Начать Kafka transaction
Вычислить и отправить выходные записи
Добавить следующие входные offsets группы в transaction
Подтвердить transaction

Producer использует transactional.id, настроенный для устойчивого владения транзакциями конкретного экземпляра приложения. Повторно запущенный владелец способен завершить судьбу старой попытки, а fencing защищает от устаревшего producer. Consumer отключает независимый auto commit. Читатели результата используют isolation.level=read_committed, чтобы пропускать отменённые транзакционные записи.

Если обработка завершилась, но transaction отменена, выход не становится доступным таким читателям и сохранённая входная позиция остаётся прежней. При повторе нужно восстановить входную позицию и заново выполнить обработку; отмена transaction сама не возвращает локальный consumer назад. Физические записи отменённой попытки могут оставаться в журнале, поэтому EOS описывает видимый результат, а не число исполнений функции.

Непрерывный read_committed также ждёт разрешения открытых транзакций у стабильной границы журнала. Долго незавершённая transaction может задержать видимость последующих записей. Транзакции дают согласованность, но требуют обработки ошибок, таймаутов и смены владельцев.

Внешний sink должен участвовать в защите

Kafka transaction не включает произвольный INSERT в PostgreSQL или HTTP-запрос к внешнему API. Для внешнего sink, приёмника, нужно спроектировать его собственный устойчивый эффект.

Первый способ — идемпотентная запись по event_id. Приёмник хранит уникальный ключ события. Повторная доставка пытается вставить существующий ключ и оставляет одну строку:

INSERT INTO event_sink(event_id, entity_id, value)
VALUES ('e-102', 17, 12500)
ON CONFLICT (event_id) DO NOTHING;

В пустом приёмнике первый запуск добавит строку, а повтор оставит одну. В практике e-102 уже сохранена, поэтому этот INSERT её не изменит. event_id должен стабильно обозначать один факт, а повтор должен иметь тот же смысл. Совпадение идентификатора с другим содержимым является ошибкой контракта, которую следует обнаружить, а не молча скрывать. Срок хранения ключей дедупликации должен покрывать период возможного replay.

Второй способ — сохранять результат и входной checkpoint одной транзакцией в той же внешней базе. При восстановлении приложение использует этот checkpoint как источник истины. Тогда повторная попытка согласуется с уже записанным результатом. Отдельный commit Kafka можно обновлять для наблюдения, но он не заменяет внешнюю атомарность.

Третий случай — обновление текущего состояния сущности. Простая перезапись безопасна для точного повтора, но старое событие после нового способно вернуть прежнее содержимое. Нужна версия: применяем событие только при incoming_version > stored_version. Если в таблице сохранена версия 3 с payload c, поздняя версия 2 с payload b должна оставить прежние версию и содержимое. Это сравнение версий, а не сортировка по времени доставки.

Практика

delivery_attempts — небольшой журнал попыток применения записей. У e-102 две попытки с числовым value = 12500, у e-103 одна с value = 7000. Сумма всех попыток равна 32000, уникальных записей — 19500. event_sink уже хранит e-102: повторная загрузка должна сохранить эту строку и добавить только e-103.

Отдельная таблица entity_state проверяет обновление версий. В ней ключ 17 имеет версию 3 с payload c, ключ 28 — версию 1 с payload x. В incoming_versions есть старые версии ключа 17, версия 2 с payload y для ключа 28 и две версии нового ключа 39: 1 с payload m, 2 с payload n.

Найди повтор, запиши каждое событие в приёмник один раз и применяй только более новые версии состояния. Каждая SQL-задача выполняется с откатом и независимо от предыдущей. Это проверка политики внешнего приёмника, а не выполнение Kafka transaction или доказательство EOS настоящего пайплайна.