Как гарантировать доставку сообщения в Kafka при сбоях и обеспечить повторную отправку

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

Понимаете ли вы, что гарантия доставки в Kafka складывается из настроек продюсера, брокера и консьюмера, и сможете ли проверить это руками. Для тестировщика следующий вопрос почти всегда такой: как вы воспроизведёте сбой и чем докажете, что сообщения не потерялись и не задвоились.

Как отвечать

Разберите цепочку по звеньям:

• продюсер: acks=all и enable.idempotence=true, чтобы ретраи не создавали дублей. В Java-клиенте с версии 3.0 это значения по умолчанию, retries почти бесконечен, а предел ретраев задаёт delivery.timeout.ms (2 минуты). В C#-клиенте Confluent.Kafka идемпотентность по умолчанию выключена, её включают EnableIdempotence = true;
• брокер и топик: replication.factor=3 и min.insync.replicas=2. Без второго параметра acks=all подтвердит запись, которая лежит на единственной живой реплике;
• консьюмер: enable.auto.commit=false и коммит оффсета после обработки. Это at-least-once, при падении сообщение придёт повторно, поэтому обработчик должен быть идемпотентным;
• повторная обработка: retry-топики с задержкой и DLQ для того, что так и не обработалось. Брокер сам в DLQ ничего не перекладывает, это делает код консьюмера, Kafka Connect или библиотека;
• база и Kafka вместе: паттерн Transactional Outbox, иначе одно из двух действий однажды потеряется.

Как тестировать: поднять три брокера в Testcontainers или docker compose, во время отправки остановить лидера партиции, через Toxiproxy оборвать сеть, убить консьюмер до коммита. После прогона сверить число отправленных, полученных и уникальных сообщений и содержимое DLQ.

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

Считают, что acks=all сам по себе гарантирует сохранность. При min.insync.replicas=1, а это значение по умолчанию, он мало отличается от acks=1.

Ставят retries=3 как «надёжную» настройку. Это ограничивает ретраи: короткий сбой, и сообщение потеряно, если ошибку в колбэке никто не обработал.

Обещают exactly-once «из коробки». Транзакции Kafka (transactional.id, isolation.level=read_committed) дают его только в цепочке «прочитал из Kafka, записал в Kafka», запись во внешнюю базу они не покрывают.