Как работает семантика Exactly-Once в Apache Kafka?

Exactly-Once Semantics (EOS) в Apache Kafka гарантирует, что каждое сообщение будет обработано потребителем ровно один раз, даже в случае сбоев продюсера, брокера или консьюмера. Это достигается комбинацией идемпотентного продюсера и транзакций.

1. Идемпотентный продюсер (Idempotent Producer):

• Цель: Предотвратить дублирование сообщений из-за повторных отправок (retries) при временных ошибках (например, таймаут сети).
• Как работает: Каждому продюсеру назначается уникальный PID (Producer ID), а каждому сообщению — монотонно растущий sequence number для каждой партиции топика. Брокер Kafka отслеживает эти номера и отбрасывает дубликаты.
• Включение: Установите enable.idempotence=true в настройках продюсера.

2. Транзакции (Transactions):

• Цель: Обеспечить атомарность при отправке сообщений в несколько партиций/топиков, а также согласованность между чтением и записью (consuming и producing) в паттерне «обработка-запись».
• Как работает: Продюсер координирует с брокером управление транзакциями. Все сообщения, отправленные между beginTransaction() и commitTransaction(), либо все будут видны консьюмерам (с isolation.level=read_committed), либо все будут отброшены.

Пример кода продюсера с Exactly-Once:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// Включаем идемпотентность и транзакции
props.put("enable.idempotence", "true"); // Неявно устанавливает acks=all, max.in.flight.requests.per.connection=5 (или 1 для <= 2.0)
props.put("transactional.id", "my-app-transactional-id"); // Должен быть уникальным и постоянным для данного продюсера

Producer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // Регистрация продюсера у координатора транзакций

try {
    producer.beginTransaction();
    // Атомарная отправка в несколько топиков
    producer.send(new ProducerRecord<>("orders", orderId, orderJson));
    producer.send(new ProducerRecord<>("audit-log", orderId, auditJson));
    // ... любая другая бизнес-логика ...
    producer.commitTransaction(); // Сообщения становятся видимыми
} catch (KafkaException e) {
    producer.abortTransaction(); // Все отправленные в транзакции сообщения отменяются
    // Обработка ошибки и возможный retry с новой транзакцией
}

Настройка консьюмера для чтения транзакционных сообщений:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// Читаем только закоммиченные транзакции
props.put("isolation.level", "read_committed");

Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("orders"));
// Консьюмер не увидит сообщения из незавершенных или прерванных транзакций.

Важные ограничения и детали:

• Требует Kafka >= 0.11.
• Накладные расходы: Транзакции добавляют задержку и используют больше памяти на брокере.
• Exactly-Once Delivery vs. Processing: Гарантия относится к доставке в Kafka. Для сквозной семантики «ровно-один-раз» в приложении (E2E) консьюмер также должен быть идемпотентным (например, сохранять смещения вместе с результатом обработки в транзакционное хранилище).
• transactional.id: Должен быть постоянным для данного логического продюсера при перезапусках, чтобы Kafka могла сопоставить его с PID и продолжить отслеживание sequence numbers.