Как добиться идемпотентности при обработке сообщений в Kafka

Что проверяют

Понимаете ли вы, что Kafka в обычной настройке доставляет сообщения at-least-once и дубли на стороне консьюмера неизбежны. Хотят услышать, откуда берутся повторы и как сделать обработку безопасной. Следом обычно спрашивают, чем идемпотентный продюсер отличается от идемпотентного консьюмера и спасают ли транзакции Kafka.

Как отвечать

Разделите задачу на две стороны.

Продюсер. Настройка enable.idempotence включает у брокера дедупликацию по идентификатору продюсера и номеру последовательности: повтор после таймаута не создаст дубль в партиции. Действует это только в пределах одной сессии продюсера и требует acks=all. В Go: в franz-go идемпотентная запись включена по умолчанию, в confluent-kafka-go (librdkafka) её включают явно.

Консьюмер. Дубли появляются при ребалансировке, при падении после обработки, но до коммита оффсета, и при повторной отправке на уровне приложения. Рабочие приёмы:

• ключ идемпотентности: в сообщении есть event_id, обработчик в одной транзакции с бизнес-изменением делает INSERT INTO processed_events ... ON CONFLICT DO NOTHING и по числу вставленных строк понимает, новое ли событие;
• естественно идемпотентные операции: UPSERT и «установить статус» вместо «прибавить к балансу»;
• версия в данных: событие применяется, только если его версия больше сохранённой, это заодно защищает от переупорядочивания;
• ручной коммит оффсетов: только после успешной обработки. В franz-go это kgo.DisableAutoCommit() и CommitRecords, в sarama MarkMessage и коммит сессии.

Транзакции Kafka дают exactly-once только для цепочки «прочитал из топика, записал в топик». Запись в PostgreSQL или HTTP-вызов в неё не входят.

Где спотыкаются

Называют acks=all способом добиться идемпотентности. Это настройка надёжности записи: сообщение не потеряется, но от дублей при обработке она не спасёт.

Проверяют «уже обработано?» отдельным SELECT, а потом пишут. Два консьюмера во время ребалансировки проходят проверку одновременно, защищает только уникальное ограничение в той же транзакции.

Коммитят оффсет до обработки. Упавший сервис тогда теряет сообщения, и это уже не дубли, а потери.