Exactly-once в Kafka: идемпотентный продюсер и транзакции

Зачем вообще нужен exactly-once

По умолчанию Kafka даёт at-least-once: продюсер может переотправить сообщение после таймаута, и в топике окажется дубликат. Для логов это терпимо, для списания денег или начисления баллов — нет. Exactly-once семантика (EOS) решает две разные проблемы: дубликаты от ретраев продюсера и консистентность записи в несколько партиций в составе одной логической операции.

Важно сразу разделить два механизма: идемпотентный продюсер (защита от дублей при ретраях) и транзакции (атомарность записи в несколько партиций/топиков, включая паттерн read-process-write).

Идемпотентный продюсер

Включается настройкой enable.idempotence=true (в новых версиях клиента — по умолчанию). Брокер выдаёт продюсеру уникальный PID (producer ID), и каждое сообщение в партицию идёт с монотонно растущим sequence number. Брокер хранит для каждого (PID, partition) последний принятый sequence number и отбрасывает дубликат, если sequence уже был обработан.

Идемпотентность работает только в рамках одной сессии продюсера и одной партиции. Если продюсер перезапустится, он получит новый PID — защита от дублей между разными запусками продюсера идемпотентностью одной не обеспечивается, для этого нужны транзакции с transactional.id.

Транзакции и transactional.id

Для атомарной записи в несколько партиций продюсер настраивается с постоянным transactional.id. Это ключ, по которому координатор транзакций (отдельный компонент на брокере) ассоциирует PID с конкретным логическим продюсером между перезапусками — то есть решает проблему "зомби"-продюсера.

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.commitTransaction();
} catch (ProducerFencedException e) {
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction();
}

Под капотом координатор ведёт внутренний топик __transaction_state, куда пишет статусы: begin, прошедшие партиции, commit/abort. При commit брокер пишет в каждую затронутую партицию служебный control-маркер (COMMIT или ABORT), который и является границей транзакции на диске.

Read-process-write: consumer + producer в одной транзакции

Самый частый сценарий EOS — Kafka Streams-подобный паттерн: читаем из топика A, обрабатываем, пишем в топик B, и коммит оффсета чтения должен быть атомарен с записью результата. Для этого продюсер коммитит не только сообщения, но и оффсеты потребителя как часть транзакции:

producer.beginTransaction();
producer.send(resultRecord);
producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata);
producer.commitTransaction();

Если процесс упадёт между отправкой результата и коммитом оффсета, при перезапуске либо вся транзакция будет видна (и оффсет, и данные), либо не будет видна ничего — потребитель перечитает тот же батч и обработает заново без дублей в выходном топике, потому что незакоммиченные записи фильтруются на чтении.

READ_COMMITTED на стороне потребителя

Записи, отправленные в рамках незакоммиченной или абортнутой транзакции, физически лежат в партиции сразу — Kafka не откатывает запись на диске. Видимость регулируется на чтении: потребитель с isolation.level=read_committed не вернёт приложению записи до тех пор, пока не увидит control-маркер commit для этой транзакции, и полностью пропустит записи с маркером abort. По умолчанию уровень — read_uncommitted, то есть все записи видны сразу, независимо от исхода транзакции.

Что это стоит на практике

EOS не бесплатен: транзакции добавляют задержку (ожидание commit перед видимостью данных), лишние round-trip к координатору, и требуют acks=all с достаточным min.insync.replicas. Для чистого логирования событий это оверхед без пользы; для финансовых операций и консистентных пайплайнов — обязательный инструмент.

Ещё разборы