Exactly-once в Apache Kafka: от проблемы дубликатов к надежной обработке

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

public class TransactionalReadProcessWrite {

    private static final String INPUT_TOPIC = "input-topic";
    private static final String OUTPUT_TOPIC = "output-topic";
    private static final String BOOTSTRAP_SERVERS = "kafka1:9092,kafka2:9092,kafka3:9092";
    private static final String CONSUMER_GROUP_ID = "my-transactional-group";
    private static final String TRANSACTIONAL_ID = "prod-cons-txn-1";

    public static void main(String[] args) {
        // Создаем продюсера и консьюмера
        KafkaProducer<String, String> producer = createProducer();
        KafkaConsumer<String, String> consumer = createConsumer();

        // Инициализируем транзакцию. Это нужно сделать один раз.
        producer.initTransactions();
        consumer.subscribe(Collections.singletonList(INPUT_TOPIC));

        System.out.println("Starting transactional processing loop...");
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                if (!records.isEmpty()) {
                    // Начинаем транзакцию
                    producer.beginTransaction();
                    try {
                        // Карта для хранения смещений
                        Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

                        for (ConsumerRecord<String, String> record : records) {
                            // 1. Обрабатываем сообщение
                            System.out.printf("Processing message: key=%s, value=%s%n", record.key(), record.value());
                            String processedValue = record.value().toUpperCase(); // Пример обработки

                            // 2. Отправляем результат в выходной топик
                            producer.send(new ProducerRecord<>(OUTPUT_TOPIC, record.key(), processedValue));
                            
                            // 3. Сохраняем смещение для коммита
                            offsetsToCommit.put(
                                new TopicPartition(record.topic(), record.partition()),
                                new OffsetAndMetadata(record.offset() + 1)
                            );
                        }

                        // 4. Коммитим смещения в рамках транзакции
                        producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

                        // 5. Завершаем транзакцию
                        producer.commitTransaction();
                        System.out.println("Transaction committed.");

                    } catch (Exception e) {
                        System.err.println("Exception occurred, aborting transaction: " + e.getMessage());
                        // 6. В случае ошибки - отменяем транзакцию
                        producer.abortTransaction();
                    }
                }
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }

    private static KafkaProducer<String, String> createProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, TRANSACTIONAL_ID);
        return new KafkaProducer<>(props);
    }

    private static KafkaConsumer<String, String> createConsumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_ID);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        return new KafkaConsumer<>(props);
    }
}

Миф о доставке и реальность обработки

Теперь, когда мы увидели механизм транзакций в действии, крайне важно прояснить один фундаментальный концептуальный момент. Часто можно услышать фразу «exactly-once delivery» (доставка ровно один раз), но в контексте распределенных систем она вводит в заблуждение. Гарантировать, что сообщение будет физически доставлено по сети и получено потребителем ровно один раз, теоретически невозможно. Эта проблема известна как «проблема двух генералов», которая доказывает, что два участника не могут со 100% уверенностью договориться о состоянии друг друга по ненадежному каналу.

Что же тогда предоставляет Apache Kafka? Kafka предлагает нечто более ценное и практически достижимое: exactly-once processing semantics (EOS), или семантику обработки ровно один раз.

Разница колоссальна:

• Exactly-once delivery (миф): Сетевой пакет с сообщением прибывает к получателю ровно один раз.
• Exactly-once processing (реальность Kafka): Эффект от обработки сообщения применяется в системе ровно один раз.

Вернемся к нашему примеру: продюсер может попытаться отправить сообщение дважды из-за сбоя сети. Однако благодаря идемпотентности, на брокере оно будет записано только один раз. Консьюмер может получить это сообщение для обработки, упасть и получить его снова после перезапуска. Но благодаря транзакциям, операция «обработать и записать результат» будет либо полностью завершена и закоммичена, либо полностью отменена. Повторная попытка просто начнется с чистого листа.

Таким образом, Kafka дает вам инструменты, чтобы гарантировать, что ваша система реагирует на сообщение ровно один раз. Это и есть истинная цель EOS.

---

 

Ограничения и подводные камни

 

Семантика exactly-once — это мощный инструмент, но не волшебная палочка. Ее гарантии имеют четкие границы, и их незнание может привести к ложному чувству безопасности.

• Границы транзакции — это границы Kafka. Гарантия EOS заканчивается там, где заканчивается контроль Kafka. Если ваш консьюмер в рамках транзакции взаимодействует с внешней системой, не поддерживающей транзакции (например, делает вызов к REST API, записывает в Redis или Memcached), вы теряете сквозную гарантию. Представьте сценарий:
  1. Ваш сервис успешно отправляет запрос во внешнюю систему (например, email-провайдеру).
  2. Сервис падает до коммита транзакции в Kafka.
  3. Kafka отменяет транзакцию. После перезапуска обработка начинается заново.
  4. Результат: Сервис отправляет во внешнюю систему повторный запрос, и пользователь получает два одинаковых письма. Для сквозного EOS внешняя система также должна быть транзакционной (например, база данных с поддержкой XA-транзакций).
• Производительность. За надежность нужно платить. Включение транзакций создает дополнительную нагрузку на брокеры и увеличивает задержку (latency). Работа координатора транзакций, двухфазный коммит и ведение логов состояния требуют ресурсов. Для систем, где критична минимальная задержка и максимальная пропускная способность, стоимость EOS может оказаться слишком высокой.
• Сложность. Код и конфигурация становятся сложнее. Вы должны тщательно управлять transactional.id, корректно обрабатывать жизненный цикл транзакции и понимать ее ограничения. Это повышает порог входа и усложняет отладку.

---

 

Лучшие практики

 

Чтобы эффективно и безопасно использовать EOS, придерживайтесь следующих правил:

Проектируйте идемпотентные консьюмеры. Даже при использовании EOS, стремитесь делать логику обработки сообщений идемпотентной, где это возможно. Это «защита в глубину». Если по какой-то причине (например, из-за ошибки в конфигурации или взаимодействия с внешней системой) дубликат все же появится, ваша система сможет его корректно обработать.

Используйте транзакционные «приемники» (sinks). Если вам нужна сквозная атомарность, используйте в качестве системы-приемника базы данных или другие системы, которые могут участвовать в общей транзакции. Для реляционных БД это обычно означает использование JDBC-коннекторов, поддерживающих стандарт XA.

Правильно управляйте transactional.id. Этот идентификатор должен быть стабильным и уникальным для каждого логического экземпляра вашего приложения-обработчика. Не генерируйте его случайным образом при каждом запуске. Стабильность transactional.id позволяет Kafka «ограждать» (fencing) устаревшие экземпляры продюсеров после сбоя, предотвращая ситуацию «split-brain».

Точно выставляйте тайм-ауты. Тайм-аут транзакции (transaction.timeout.ms) должен быть настроен с учетом максимального времени, которое может потребоваться на обработку батча сообщений. Он должен быть меньше, чем тайм-аут сессии консьюмера (max.poll.interval.ms), чтобы избежать нежелательных ребалансировок группы.

---

 

Мониторинг и отладка

 

Работающие в режиме EOS системы требуют особого внимания к мониторингу.

Ключевые метрики: Настройте сбор и оповещения для JMX-метрик с брокеров и клиентов.

На брокере (Transaction Coordinator):

kafka.server:type=transaction-coordinator-metrics,name=transaction-commits-total

kafka.server:type=transaction-coordinator-metrics,name=transaction-aborts-total

Резкий рост числа transaction-aborts-total может свидетельствовать о проблемах в приложении-обработчике.

На продюсере:

Следите за метриками, связанными со статусом транзакций и ошибками.

Распространенные исключения: Будьте готовы к отладке специфичных для EOS ошибок.

ProducerFencedException: Это не ошибка, а штатный механизм защиты. Он возникает, когда запускается новый экземпляр продюсера с тем же transactional.id, что и у активного. Kafka «ограждает» старый экземпляр, чтобы только новый мог продолжать работу. Вы должны корректно завершать работу старого экземпляра при получении этого исключения.

OutOfOrderSequenceException: Указывает на то, что брокер получил сообщение с некорректным порядковым номером. Это может означать серьезную проблему или потерю данных, требующую расследования.

InvalidProducerEpochException: Означает, что эпоха продюсера устарела. Обычно это происходит после ограждения (fencing).

---

 

Заключение

 

Семантика exactly-once в Apache Kafka — это одна из самых мощных и в то же время неправильно понимаемых возможностей платформы. Она не предлагает мифическую «доставку ровно один раз», а предоставляет инженерам надежный и прагматичный инструментарий для реализации обработки ровно один раз.

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

Однако эта мощь требует дисциплины. EOS — это не «серебряная пуля», а компромисс между надежностью, производительностью и сложностью. Прежде чем внедрять его, всегда задавайте себе вопрос: действительно ли мой бизнес-кейс не может пережить дубликаты, которые можно отловить на уровне приложения? Часто оказывается, что хорошо спроектированная система с at-least-once и идемпотентной логикой является более простым и эффективным решением. Начинайте с него и переходите к EOS только тогда, когда требования к консистентности делают этот шаг абсолютно необходимым.