📨 Розділ 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 вичитує ці блоки мережею, звіряє їх зі списком скасованих транзакцій (збережених у пам’яті) і відкидає їх на стороні клієнта, витрачаючи мережу і 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() потрібно викликати перед кожним відправленим повідомленням» — він викликається суворо один раз при старті життєвого циклу продюсера.

Пов’язані запитання