📨 Раздел 15 · Вопрос #11

Как настроить exactly-once семантику

Для настройки семантики Exactly-Once (EOS) в Apache Kafka необходимо согласованно настроить все три звена системы: продюсера, брокеры и консьюмера, а также использовать специаль...

📨 Kafka 📄 Исходный Markdown

🟢 Junior Level

30-секундный ответ

Для настройки семантики Exactly-Once (EOS) в Apache Kafka необходимо согласованно настроить все три звена системы: продюсера, брокеры и консьюмера, а также использовать специальный Transaction API:

  1. Продюсер:
    • Задать уникальный постоянный transactional.id.
    • Включить enable.idempotence = true (дефолт в Java 3.0+).
    • Выставить acks = all и retries = Integer.MAX_VALUE.
  2. Брокер:
    • Гарантировать репликацию журнала транзакций: transaction.state.log.replication.factor = 3 и transaction.state.log.min.isr = 2.
  3. Консьюмер:
    • Выставить уровень изоляции: isolation.level = read_committed.
    • Отключить автокоммит: enable.auto.commit = false.
  4. Код (Read-Process-Write):
    • Выполнять чтение, обработку, запись в выходной топик и фиксацию оффсетов консьюмера в рамках единой транзакции через producer.sendOffsetsToTransaction().

Наглядная аналогия из жизни

Представьте бухгалтерский перевод средств:

  • Вы переводите деньги с одного банковского счёта на другой: нужно списать сумму с расчётного счёта А и зачислить на расчётный счёт Б.
  • Если снять деньги со счёта А, но сервер банка упадёт до зачисления на счёт Б — деньги исчезнут.
  • Если зачислить на счёт Б, но упасть до списания с А — возникнут бесплатные деньги из воздуха.
  • Транзакция (Exactly-Once) гарантирует: либо обе записи появляются в главной бухгалтерской книге одновременно, либо обе отменяются, возвращая систему к исходному балансу.

Схема топологии Read-Process-Write

             ┌────────────────────────────────────────────────────────┐
             │               ЕДИНАЯ АТОМАРНАЯ ТРАНЗАКЦИЯ              │
             │                                                        │
[ Топик A ] ─┼─► [ Consumer.poll() ]                                  │
             │            │                                           │
             │            ▼                                           │
             │      [ Обработка ]                                     │
             │            │                                           │
             │            ▼                                           │
             │   [ Producer.send() ] ────────────────► [ Топик B ]    │
             │            │                                           │
             │            ▼                                           │
             │ [ sendOffsetsToTransaction() ] ───────► [ __consumer_offsets ]
             └────────────────────────────────────────────────────────┘

🟡 Middle Level

Реализация Read-Process-Write через чистый Java API

В чистом Java SDK для согласования вычитки и отправки используется метод sendOffsetsToTransaction():

Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-processor-tx-1");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor-group");
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

// 1. Инициализация продюсера (вызывается строго 1 раз при старте приложения!)
producer.initTransactions();
consumer.subscribe(List.of("input-orders"));

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;

    try {
        // 2. Начало транзакции
        producer.beginTransaction();

        Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();

        for (ConsumerRecord<String, String> record : records) {
            String processedValue = enrichOrder(record.value());
            producer.send(new ProducerRecord<>("output-orders", record.key(), processedValue));

            // Сохраняем следующее смещение для коммита
            offsetsToCommit.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
        }

        // 3. Привязка оффсетов консьюмера к транзакции продюсера (KIP-447)
        producer.sendOffsetsToTransaction(offsetsToCommit, consumer.groupMetadata());

        // 4. Атомарная фиксация (коммитят и сообщения, и оффсеты)
        producer.commitTransaction();

    } catch (ProducerFencedException | InvalidProducerEpochException e) {
        log.error("Продюсер был вытеснен новым инстансом (Zombie). Завершаем работу.", e);
        producer.close();
        break;
    } catch (KafkaException e) {
        log.warn("Ошибка выполнения транзакции. Откатываемся...", e);
        // 5. Откат транзакции при любых сетевых или программных сбоях
        producer.abortTransaction();
        // Сбрасываем позицию консьюмера на последний закоммиченный оффсет
        resetToLastCommitted(consumer);
    }
}

Настройка в Spring Boot 3 (KafkaTransactionManager)

В Spring Boot настройка EOS сводится к объявлению транзакционного менеджера и аннотации @Transactional:

@Configuration
public class KafkaTxConfig {

    @Bean
    public DefaultKafkaProducerFactory<String, Object> producerFactory() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configs.put(ProducerConfig.ACKS_CONFIG, "all");
        configs.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        
        DefaultKafkaProducerFactory<String, Object> factory = new DefaultKafkaProducerFactory<>(configs);
        // Задание префикса включает режим транзакций
        factory.setTransactionIdPrefix("tx-order-");
        return factory;
    }

    @Bean
    public KafkaTransactionManager<String, Object> kafkaTransactionManager(
            ProducerFactory<String, Object> producerFactory) {
        return new KafkaTransactionManager<>(producerFactory);
    }
}

@Service
@RequiredArgsConstructor
public class TransactionalOrderListener {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    @Transactional("kafkaTransactionManager")
    @KafkaListener(topics = "input-orders", groupId = "spring-tx-group",
                   containerFactory = "kafkaListenerContainerFactory")
    public void process(ConsumerRecord<String, OrderDto> record) {
        OrderDto order = record.value();
        ProcessedOrder result = transform(order);
        
        // Отправка в выходной топик внутри открытой транзакции
        kafkaTemplate.send("output-orders", record.key(), result);
        
        // Оффсет консьюмера будет автоматически отправлен в транзакцию при выходе из метода
    }
}

🔴 Senior Level

Защита от зомби через Producer Epoch (Zombie Fencing)

В распределенной системе под с продюсером может зависнуть в долгом GC (Stop-the-World pause). Kubernetes решит, что под умер, и поднимет дубликат на соседней ноде. Если старый под очнется, возникнет ситуация двух активных продюсеров с одинаковым назначением (Split-Brain).

Как Kafka решает эту проблему через transactional.id:

  1. При старте новый инстанс продюсера с тем же transactional.id вызывает initTransactions().
  2. Брокер Transaction Coordinator увеличивает внутренний счетчик эпохи: Producer Epoch = Epoch + 1.
  3. Координатор немедленно и принудительно отменяет (ABORT) все незавершенные транзакции старой эпохи.
  4. Когда «зомби»-под отвисает и пытается вызвать commitTransaction() со старой эпохой, координатор возвращает ошибку ProducerFencedException.
  5. Зомби-процесс получает фатальное исключение и аварийно завершает работу, не успев испортить данные.

Эволюция EOS: KIP-447 (EOS v2 в Kafka 2.5+)

  • EOS v1 (до Kafka 2.5): требовал отдельного пула транзакционных продюсеров для каждого набора партиций консьюмера. Если при ребалансе менялось назначение партиций, транзакции зависали или требовали перезапуска всего контекста.
  • EOS v2 (KIP-447): метод producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata()) принимает метаданные группы консьюмеров (включая текущий generation.id и member.id). Брокер проверяет оффсеты прямо в координаторе транзакций, предотвращая коммит оффсетов, если произошел ребаланс группы.
  • В Kafka Streams: достаточно выставить одну строчку конфигурации: 
    processing.guarantee=exactly_once_v2

4 Tricky Questions

1. Зачем в метод sendOffsetsToTransaction() начиная с Kafka 2.5 передается аргумент consumer.groupMetadata(), и какую уязвимость это закрыло?

Ответ:

  • В ранних версиях (EOS v1) sendOffsetsToTransaction() принимал только мапу смещений и groupId.
  • Уязвимость: если у консьюмера завис поток, а координатор группы исключил его по max.poll.interval.ms и перераспределил партицию другому узлу, старый консьюмер мог после отвисания закоммитить оффсеты через продюсера в рамках транзакции. Транзакционный координатор не знал, что консьюмер уже исключен из группы, и принимал оффсеты, перетирая позицию нового владельца партиции!
  • ConsumerGroupMetadata передает брокеру актуальные generationId, memberId и groupInstanceId.
  • Координатор транзакций валидирует эти метаданные у координатора группы: если поколение консьюмера устарело, транзакция сбрасывается с ошибкой CommitFailedException или FencedInstanceIdException, предотвращая рассинхронизацию группы.

2. Что произойдет, если в кластере Kafka параметр брокера transaction.state.log.replication.factor оставлен равным 1, а нода с координатором транзакций вышла из строя?

Ответ:

  • Топик __transaction_state хранит текущее состояние всех активных транзакций и соответствие transactional.id -> PID / Epoch.
  • Если фактор репликации равен 1, то при физической аварии брокера, на котором размещена партиция этого топика, метаданные транзакций становятся недоступными.
  • Последствия:
    • Все продюсеры, чьи транзакции координировались этим брокером, получат критическую ошибку CoordinatorNotAvailableException или TimeoutException.
    • Ни закоммитить, ни откатить открытые транзакции невозможно.
    • Консьюмеры с read_committed на целевых топиках намертво заблокируются на смещении LSO (Last Stable Offset) и не смогут читать даже новые сообщения от других исправных продюсеров!
  • Требование для Production: transaction.state.log.replication.factor = 3 и transaction.state.log.min.isr = 2.

3. Почему разработчикам категорически не рекомендуется выставлять слишком большой transaction.timeout.ms (например, 1 час)?

Ответ:

  • Консьюмеры в режиме read_committed вычитывают сообщения строго до отметки LSO (Last Stable Offset) — смещения первой открытой транзакции.
  • Если транзакционный продюсер открыл транзакцию и аварийно завис (out of memory, мертвая блокировка потоков), брокер держит эту транзакцию в статусе ONGOING вплоть до истечения transaction.timeout.ms.
  • Всё это время LSO не сдвигается! Ни один консьюмер кластера с read_committed не получит ни одного сообщения из этой партиции, даже если параллельно другие продюсеры закоммитят миллионы записей.
  • Downstream-сервисы получат катастрофический рост consumer lag. Рекомендуемое значение таймаута транзакции — от 10 до 60 секунд (transaction.timeout.ms=30000).

4. Может ли возникнуть утечка памяти (Memory Leak) на брокере или в топике из-за постоянных откатов (abortTransaction())?

Ответ:

  • Нет, утечки памяти на JVM брокера не происходит, но возникает дисковое замусоривание (Log Bloat).
  • При вызове abortTransaction() физические сообщения из файлов сегментов лога .log не удаляются! Брокер лишь дописывает вслед за ними служебный маркер управления ABORT.
  • Консьюмер read_committed вычитывает эти блоки по сети, сверяет их со списком отмененных транзакций (stored in memory) и отбрасывает их на стороне клиента, расходуя сеть и CPU.
  • Отмененные сообщения будут удалены с диска только тогда, когда сегмент целиком устареет по времени (retention.ms) или размеру (retention.bytes) и будет удален сборщиком логов брокера.

🎯 Шпаргалка для интервью

Главное о настройке Exactly-Once

  • Чеклист продюсера: transactional.id, enable.idempotence=true, acks=all.
  • Чеклист консьюмера: isolation.level=read_committed, enable.auto.commit=false.
  • Чеклист брокера: transaction.state.log.replication.factor=3, transaction.state.log.min.isr=2.
  • Read-Process-Write: оффсеты консьюмера коммитятся через producer.sendOffsetsToTransaction(offsets, groupMetadata).
  • Zombie Fencing: Producer Epoch делает невозможной одновременную работу старого и нового пода под одним transactional.id.
  • LSO блокировка: пока транзакция не завершена (commit или abort), консьюмеры read_committed не увидят идущие за ней данные.
  • Kafka Streams: processing.guarantee=exactly_once_v2 берет всю координацию на себя из коробки.

Сравнение поведения продюсеров

| Параметр | Идемпотентный Producer | Транзакционный Producer | | :— | :— | :— | | Идентификатор | Динамический PID (меняется при рестарте) | Постоянный transactional.id (сохраняется) | | Область действия | 1 партиция одного топика | Множество топиков + __consumer_offsets | | Защита от сбоя | Защита от дублей при сетевом retry | Атомарный мультипартиционный коммит/откат | | Сложность настройки | Включен по умолчанию | Требует ручной координации кода или Spring |

Красные флаги на собеседовании (Чего говорить нельзя)

  • ❌ «Для exactly-once достаточно просто включить idempotence на продюсере» — идемпотентность защищает только одну партицию при ретраях, для атомарности нескольких партиций и оффсетов нужны транзакции.
  • ❌ «Консьюмер в транзакции коммитит свои смещения через consumer.commitSync()» — это ломает атомарность! Оффсеты обязаны отправляться через producer.sendOffsetsToTransaction().
  • ❌ «initTransactions() нужно вызывать перед каждым отправленным сообщением» — он вызывается строго один раз при старте жизненного цикла продюсера.

Связанные вопросы