Що таке idempotent producer

У розподілених системах втрата мережевих пакетів — звичайне явище. Без механізму ідемпотентності виникає класична проблема дублювання даних:

🟢 Junior Level

Що таке Idempotent Producer простими словами

Idempotent Producer (ідемпотентний продюсер) в Apache Kafka — це режим роботи клієнта відправки повідомлень, який математично гарантує, що кожен запис буде збережено в партиції брокера рівно один раз (exactly-once write per partition), навіть якщо через мережеві збої продюсер здійснює повторні спроби відправки (retries).

Сценарій БЕЗ ідемпотентності:
1. Producer ──send(Message A)──> Broker (Записав у лог offset 42)
2. Broker   ──ACK (TCP)──X (Мережевий збій / втрата пакета ACK)
3. Producer (таймаут) ──retry(Message A)──> Broker (Записав у лог offset 43)
Підсумок: Повідомлення A збережено двічі! Виникає дублікат бізнес-події (подвійне списання грошей, створення двох замовлень).

Сценарій З ідемпотентністю (enable.idempotence = true):
1. Producer ──send(PID=101, Seq=0, Message A)──> Broker (Записав у лог offset 42, запам'ятав Seq=0)
2. Broker   ──ACK (TCP)──X (Втрата пакета ACK)
3. Producer (таймаут) ──retry(PID=101, Seq=0, Message A)──> Broker
4. Broker бачить: для PID=101 запис із Seq=0 вже існує!
   -> Брокер відхиляє дублюючий запис на диск, але відправляє Producer успішний ACK(offset 42).
Підсумок: У лозі рівно один запис. Дублікатів немає!

Як увімкнути ідемпотентність у Java

Починаючи з Kafka 3.0, ідемпотентність продюсера увімкнена за замовчуванням (enable.idempotence = true). Для версій Kafka 0.11–2.8 її потрібно було вказувати явно:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092,kafka-broker2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// Увімкнення ідемпотентності (за замовчуванням true в Kafka 3.0+)
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

// Створення клієнта
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

Ключові ідентифікатори

Дедуплікація спирається на дві службові мітки, які клієнт прозоро додає до повідомлень:

Producer ID (PID): 64-бітний унікальний ідентифікатор, що автоматично видається брокером продюсеру під час ініціалізації клієнта.
Sequence Number (SeqNum): 32-бітний монотонно зростаючий лічильник ($0, 1, 2, \dots$), прив’язаний до пари (PID, TopicPartition).

🟡 Middle Level

Неявні (автоматичні) налаштування при `enable.idempotence=true`

Коли властивість enable.idempotence встановлена в true, клієнт Kafka Producer автоматично валідує і виставляє обов’язкові залежні параметри:

acks = all (acks = -1): Гарантія запису на всі репліки з ISR (In-Sync Replicas). Без підтвердження від усіх реплік брокер не може гарантувати цілісність порядку номерів послідовності при можливій зміні лідера (Leader Election).
retries = Integer.MAX_VALUE: Кількість спроб повторної відправки стає фактично нескінченною (обмежується лише таймаутом delivery.timeout.ms). Продюсер не здається при тимчасових мережевих помилках, оскільки повтори тепер абсолютно безпечні.
max.in.flight.requests.per.connection <= 5: Кількість непідтверджених мережевих запитів, що одночасно знаходяться в сокеті до одного брокера. Значення за замовчуванням — 5. Брокер здатний коректно дедуплікувати та впорядковувати до 5 паралельних запитів (починаючи з Kafka 1.1).

Небезпечна помилка конфігурації: Якщо розробник явно задасть enable.idempotence=true і одночасно вкаже acks=1 або max.in.flight.requests.per.connection=10, конструктор KafkaProducer викине ConfigException: org.apache.kafka.common.config.ConfigException: Must set acks to all when enable.idempotence is true.

Порівняння рівнів гарантій доставки

Режим продюсера	Конфігурація	Гарантія	Поведінка при збої мережі	Ризик дублікатів
At-Most-Once	`acks=0` або `retries=0`	Повідомлення доставляється максимум один раз	Повідомлення втрачається назавжди	Нульовий (немає повторів)
At-Least-Once	`acks=1/all`, `retries>0`, `idempotence=false`	Повідомлення доставляється мінімум один раз	Повторна відправка створює дублікати	Високий
Idempotent (At-Least-Once + Deduplication)	`enable.idempotence=true`	Рівно один екземпляр у партиції	Повторна відправка дедуплікується брокером	Нульовий (у межах сесії)
Exactly-Once Semantics (EOS)	Idempotence + `transactional.id` + Transactions API	Атомарний запис у декілька топіків/партицій	Транзакція відкочується або фіксується цілком	Нульовий (наскрізний end-to-end)

Межі можливостей Idempotent Producer

Хоча ідемпотентний продюсер повністю вирішує проблему мережевих дублікатів, він має суворі фундаментальні обмеження:

Тільки в межах однієї партиції: Лічильник SequenceNumber ведеться незалежно для кожної TopicPartition. Дедуплікація між різними партиціями або різними топіками не працює.
Тільки в межах однієї сесії продюсера (Session-Scoped): PID зберігається виключно в пам’яті процесу KafkaProducer. Якщо JVM перезапустилася (crash, redeploy), новий екземпляр продюсера запитає у брокера новий PID. Якщо попередній процес встиг відправити повідомлення, але впав до обробки відповіді, а новий процес повторить бізнес-операцію — брокер вважатиме це повідомленням від нового продюсера і збереже дублікат.
Не захищає від логічних дублікатів застосунку: Якщо зовнішній користувач натиснув кнопку «Оплатити замовлення» двічі в інтерфейсі браузера, і бекенд породив два окремі виклики producer.send() з новими об’єктами — для Kafka це два різних повідомлення з різними Sequence Numbers.

🔴 Senior Level

Архітектурна будова: KIP-98 та ProducerStateManager

На брокері логіка перевірки ідемпотентності реалізована компонентом ProducerStateManager (пакет kafka.log).

                Producer RecordBatch (PID: 5042, Epoch: 0, BaseSeq: 10, LastSeq: 14)
                                             │
                                             ▼
                             Брокер: Partition Replica
                                             │
                                             ▼
                                   ProducerStateManager
         ┌────────────────────────────────────────────────────────────────────────┐
         │ Карта станів у пам'яті: Map<Long, ProducerStateEntry>                 │
         │ Key: PID (5042)                                                        │
         │ Value:                                                                 │
         │   - ProducerEpoch: 0                                                   │
         │   - LastSequence: 9 (Очікуваний наступний: 10)                        │
         │   - Sliding Queue останніх 5 батчів: [Seq 5, Seq 6, Seq 7, Seq 8, Seq 9]│
         │   - LastUpdatedTimestamp: 1711456000000                                │
         └────────────────────────────────────────────────────────────────────────┘
                                             │
                  ┌──────────────────────────┴──────────────────────────┐
                  ▼                                                     ▼
     BaseSeq == LastSequence + 1 (10 == 9 + 1)             BaseSeq <= LastSequence (10 <= 9)
                  │                                                     │
                  ▼                                                     ▼
          [ВАЛІДНИЙ ПАКЕТ]                                      [ДУБЛІКАТ]
  - Запис в Active Segment File (.log)                - Запис на диск НЕ виконується
  - Оновлення LastSequence = 14                       - Формується успішний ACK
  - Збереження знімка у *.snapshot файл                 з вихідним baseOffset

Ініціалізація Producer ID (`InitProducerIdRequest`)

Коли продюсер стартує:

Він відправляє InitProducerIdRequest брокеру (в ролі координатора транзакцій Transaction Coordinator).
Брокер генерує монотонно зростаючий 64-бітний ProducerId та 16-бітний ProducerEpoch (початкове значення 0).
Присвоєний PID закріплюється за продюсером.

Формат пакета RecordBatch (KIP-98)

Кожен надісланий пакет RecordBatch містить у заголовку:

ProducerId (int64)
ProducerEpoch (int16)
BaseSequence (int32) — номер послідовності першого повідомлення в даному батчі.

Усередині батча кожне повідомлення не зберігає повний лічильник, а має лише відносний зсув: offsetDelta та sequenceDelta ($0, 1, 2, \dots, N-1$). Це мінімізує накладні витрати на байти в мережі.

Логіка перевірки на брокері

Коли RecordBatch надходить на брокер у лідер-партицію, ProducerStateManager звіряє вхідний BaseSequence з LastSequence для даного PID:

$BaseSequence = LastSequence + 1$ (Ідеальний порядок): Батч валідний. Брокер апендить записи в активний сегмент лога .log, оновлює LastSequence на значення LastSequence = BaseSequence + recordCount - 1 і повертає RecordMetadata із призначеними офсетами.
$BaseSequence \le LastSequence$ (Дублікат): Брокер визначає, що батч із цими номерами вже було записано в лог. Батч ігнорується (дискового вводу-виводу немає). Брокер повертає клієнту штатний ACK з тим самим зсувом (offset), який було присвоєно оригінальному батчу. Продюсер вважає відправку успішною.
$BaseSequence > LastSequence + 1$ (Порушення порядку / втрата даних): Брокер виявляє «дірку» в послідовності (наприклад, очікував Seq 10, а прийшов Seq 12). Це означає, що проміжний пакет Seq 10-11 було втрачено або він застряг. Брокер відхиляє запит фатальною помилкою OutOfOrderSequenceException.

Механізм ковзного вікна: чому допустимий `max.in.flight.requests.per.connection = 5`

Історично (в Kafka 0.11) для роботи ідемпотентності вимагалося виставляти max.in.flight.requests.per.connection = 1. Якщо надсилалося декілька запитів паралельно (R1, R2), і R1 падав із мережевою помилкою, а R2 долітав успішно, відбувалося порушення порядку на диску.

У Kafka 1.1 (KIP-180) брокер навчили керувати чергою з 5 непідтверджених запитів:

Брокер утримує чергу вхідних батчів для кожного PID.
Якщо запит $R_2$ (Seq 5–9) прибув на брокер раніше, ніж завершився повтор запиту $R_1$ (Seq 0–4), брокер тимчасово не апендить $R_2$ у лог, а очікує прибуття $R_1$.
Щойно $R_1$ успішно записано, $R_2$ негайно записується слідом без генерації OutOfOrderSequenceException.
Якщо розмір вікна перевищує 5, брокеру не вистачає метаданих для вирішення перестановок, що й зумовлює жорсткий ліміт $\le 5$.

Персистентність станів та перемикання лідерів: `.snapshot` файли

Що відбувається, якщо брокер перезапустився або змінився лідер партиції?

У каталозі партиції поруч із файлами сегментів лога (.log) та індексами (.index, .timeindex) Kafka зберігає файли знімків станів продюсерів: 00000000000000010000.snapshot.
У snapshot зберігається серіалізований стан ProducerStateManager (активні PID, їхні епохи та останні 5 послідовностей) на момент створення кожного сегмента лога.
Фоновий процес реплікації копіює записи зі збереженням усіх заголовків RecordBatch (включаючи PID і SeqNum).
Follower-репліки підтримують власний синхронний ProducerStateManager. При зміні лідера новий лідер миттєво готовий продовжувати перевірку ідемпотентності без втрати контексту.
Записи про неактивних продюсерів автоматично видаляються брокером після спливання transactional.id.expiration.ms / producer.id.expiration.ms (за замовчуванням 7 днів).

4 Tricky Questions

1. Чи може виклик `producer.send()` з `enable.idempotence=true` завершитися винятком `DuplicateSequenceException` на клієнті?

Відповідь: Ні, клієнт KafkaProducer ніколи не викидає DuplicateSequenceException у прикладний код. Коли брокер виявляє дублюючий батч ($BaseSequence \le LastSequence$), він трактує це як штатний мережевий retry після успішного запису. Брокер відкидає тіло повідомлення, знаходить вихідний присвоєний offset і відправляє продюсеру стандартний ProduceResponse з ознакою успіху. Для клієнтського застосунку виклик callback.onCompletion(metadata, null) або future.get() завершується абсолютно прозоро й успішно.

2. Чому перезапуск пода (Pod Restart) у Kubernetes ламає гарантію ідемпотентності, і як досягти гарантії після рестарту?

Відповідь:

При перезапуску застосунку (Pod Restart / OOM / Deployment) екземпляр KafkaProducer знищується.
Новий створений інстанс продюсера виконує InitProducerIdRequest і отримує абсолютно новий випадковий PID.
Брокер асоціює новий PID з нульовим SequenceNumber і нічого не знає про зв’язок цього інстанса з попереднім загиблим процесом.
Якщо впалий процес встиг відправити батч у брокер, але не встиг зафіксувати це в локальній БД або зовнішній черзі, новий процес може повторно сформувати ту саму бізнес-подію. Для брокера це будуть різні PID, і повідомлення запишеться вдруге.
Рішення: Для дедуплікації через рестарти застосунку необхідний Transactional Producer: задається стабільний строковий ідентифікатор props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-service-instance-1"). При старті з тим самим transactional.id координатор брокера знаходить старий PID, інкрементує ProducerEpoch (відсікаючи зомбі-інстанси через ProducerFencedException) і зберігає безперервну транзакційну історію.

3. Що станеться, якщо продюсер спробує відправити батч з `max.in.flight.requests.per.connection = 7` при увімкненій ідемпотентності?

Відповідь: Поведінка залежить від версії клієнтської бібліотеки Kafka:

Починаючи з версії Kafka 2.0+, спроба ініціалізувати клієнт з enable.idempotence=true та max.in.flight.requests.per.connection > 5 призведе до негайної помилки конфігурації при старті застосунку: ConfigException: Must set max.in.flight.requests.per.connection to at most 5 to use the idempotent producer.
Це обмеження зумовлене тим, що структура дедуплікації ProducerStateManager на брокері виділяє рівно 5 слотів для зберігання останніх порядкових номерів для кожного з’єднання. Перевищення цього ліміту призвело б до хибних OutOfOrderSequenceException у разі мережевого пересортування пакетів.

4. Чому при увімкненій ідемпотентності критично важливо зберігати `acks=all`, а не `acks=1`?

Відповідь: Уявімо сценарій з acks=1:

Producer відправляє батч (PID 100, Seq 1) на брокер-лідер.
Лідер записує батч на свій диск і одразу шле ACK продюсеру, не чекаючи реплікації на followers.
Продюсер отримує ACK і збільшує локальний лічильник до Seq 2.
У цей момент лідер фізично згорає (hardware crash).
Follower обирається новим лідером. Але він не встиг отримати запис Seq 1! У його ProducerStateManager останнім відомим записом є Seq 0.
Продюсер відправляє наступний пакет (PID 100, Seq 2).
Новий лідер бачить: $BaseSequence (2) > LastSequence (0) + 1$.
Відбувається фатальний збій: брокер повертає помилку OutOfOrderSequenceException. Продюсер блокується і більше не може надсилати повідомлення в дану партицію без перезапуску. Тому acks=all є суворою математичною вимогою консистентності розподіленого стану реплік.

🎯 Шпаргалка для інтерв’ю

30-секундна відповідь (Elevator Pitch)

«Idempotent Producer — це механізм Kafka (увімкнений за замовчуванням з версії 3.0), який виключає появу дублікатів повідомлень у партиціях при мережевих повторах (retries). При старті брокер присвоює продюсеру унікальний ідентифікатор ProducerId (PID), а клієнт нумерує кожне повідомлення монотонним лічильником SequenceNumber у розрізі кожної партиції. Брокер валідує послідовність через ProducerStateManager. Якщо через втрату ACK продюсер повторно шле вже записаний батч, брокер відкидає дублікат, але повертає клієнту успішний ACK. Увімкнення ідемпотентності автоматично форсує acks=all, retries=INT_MAX та max.in.flight.requests.per.connection <= 5. Ідемпотентність працює суворо в межах однієї партиції та однієї сесії процесу продюсера».

Ключові метрики для Production-моніторингу

kafka.producer:type=producer-metrics,client-id={id} -> record-error-rate: частота помилок відправки повідомлень.
kafka.producer:type=producer-metrics,client-id={id} -> record-retry-rate: частота повторних спроб відправки. Якщо retry-rate високий, але дублікатів у топіку немає — ідемпотентність успішно відпрацьовує.
kafka.network:type=RequestMetrics,name=ErrorsPerSec,request=Produce,error=OutOfOrderSequenceException: наявність помилок розсинхронізації послідовностей на брокерах (сигнал про серйозні мережеві проблеми або нештатні збої клієнтів).

Червоні прапорці (Чого говорити НЕ МОЖНА)

❌ «Ідемпотентний продюсер гарантує дедуплікацію у всьому кластері між усіма сервісами» — Дедуплікація працює суворо в розрізі однієї партиції та одного конкретного інстанса продюсера.
❌ «Якщо впав сервіс, після рестарту ідемпотентний продюсер не допустить дублікатів» — Допустить, оскільки новий процес отримає новий PID. Для дедуплікації через рестарти потрібен transactional.id.
❌ «Ідемпотентний продюсер вимагає max.in.flight.requests.per.connection = 1» — Застаріло. Починаючи з Kafka 1.1 допустиме значення до 5 включно.
❌ «При виявленні дубліката брокер викидає помилку продюсеру» — Брокер тихо відкидає дублікат і відповідає штатним ACK про успіх.

🟢 Junior Level