📨 Розділ 15 · Питання #13

Як працює commit offset

Ключові моменти:

📨 Kafka 📄 Вихідний Markdown

🟢 Junior Level

30-секундна відповідь

Commit offset (фіксація зміщення) — це операція, при якій консьюмер повідомляє кластеру Kafka: «Наша група консьюмерів успішно прийняла та обробила повідомлення партиції $P$ аж до зміщення $N-1$. При перезапуску або ребалансуванні групи передайте нам дані починаючи зі зміщення $N$».

Ключові моменти:

  • Золоте правило +1: у Kafka фіксується не номер поточного обробленого повідомлення, а зміщення наступного очікуваного повідомлення (record.offset() + 1).
  • Куди пишеться: дані зберігаються у внутрішній системний топік __consumer_offsets.
  • Навіщо потрібен: без коміту офсетів сервіс, що впав, при перезапуску почне читати всі повідомлення заново з початку топіка або з моменту старту групи, породжуючи масові дублікати.

Наочна аналогія з життя

Уявіть закладку в книзі рецептів:

  • Кухар готує страви по черзі: рецепт №1, №2, №3.
  • Приготувавши страву №3, кухар перекладає закладку на рецепт №4 («№3 я вже зробив, завтра почну з №4»).
  • Якщо посеред готування на кухні згасне світло, вранці кухар відкриє книгу на рецепті №4 і не буде готувати перші три страви заново.

Візуалізація правила +1 при коміті

Партиція топіка:
[ Offset 0 ] [ Offset 1 ] [ Offset 2 ] [ Offset 3 ] [ Offset 4 ]
                              ▲              ▲
                              │              │
                    Оброблений запис         │
                    record.offset() = 2      │
                                             │
                              Фіксований Committed Offset
                              (record.offset() + 1 = 3)

🟡 Middle Level

Синхронний vs Асинхронний коміт

У клієнті KafkaConsumer доступні два базові методи ручної фіксації:

// 1. Синхронний коміт (commitSync)
// Блокує потік до отримання підтвердження від брокера (5-50 мс)
// Автоматично робить retry при тимчасових мережевих збоях
consumer.commitSync();

// 2. Асинхронний коміт (commitAsync)
// Не блокує потік, надсилає мережевий запит у фоновому режимі (<1 мс)
// НІКОЛИ не робить retry при збоях задля уникнення гонок версій офсетів!
consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        log.warn("Асинхронний коміт не вдався для зміщень {}", offsets, exception);
    }
});

Патерн Production-обробки: Async у циклі + Sync при завершенні

Щоб не гальмувати цикл обробки синхронними очікуваннями, але гарантувати збереження позиції при зупинці сервісу, застосовують наступний патерн:

try {
    while (running) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            processBusinessLogic(record);
        }
        // Швидкий неблокуючий коміт в основному циклі
        consumer.commitAsync();
    }
} catch (WakeupException e) {
    // Штатна зупинка консьюмера
} catch (Exception e) {
    log.error("Непередбачена помилка в циклі обробки", e);
} finally {
    try {
        // Фінальний синхронний коміт перед закриттям для фіксації останніх даних
        consumer.commitSync();
    } finally {
        consumer.close();
    }
}

Точковий коміт зміщень (Per-Partition Commit)

Якщо батч містить тисячі записів із різних партицій, можна комітити прогрес порційно:

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

for (ConsumerRecord<String, String> record : records) {
    process(record);

    currentOffsets.put(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1, "metadata-audit-v1")
    );

    if (processedCount % 100 == 0) {
        consumer.commitAsync(currentOffsets, null);
    }
}

🔴 Senior Level

Мережевий протокол і внутрішня будова коміту

  1. Відправлення запиту: Консьюмер формує мережевий пакет OffsetCommitRequest і відправляє його Group Coordinator (брокеру, що керує даною групою).
  2. Запис у __consumer_offsets:
    • Координатор отримує запит і формує повідомлення для топіка __consumer_offsets.
    • Ключ повідомлення: [Group ID, Topic, Partition].
    • Значення: [Committed Offset, Leader Epoch, Metadata, Commit Timestamp].
    • Координатор пише запис у свій локальний журнал і чекає його реплікації на кворум реплік відповідно до налаштування offsets.topic.replication.factor.
  3. Кеш у пам’яті: Після успішної реплікації координатор оновлює свій внутрішній in-memory offset cache. Усі подальші запити OffsetFetchRequest віддаються миттєво з оперативної пам’яті брокера без звернення до диска.
  4. Відповідь клієнту: Брокер повертає клієнту пакет OffsetCommitResponse.

Причини виникнення CommitFailedException

Одна з найпоширеніших помилок у продакшені:

org.apache.kafka.clients.consumer.CommitFailedException: 
Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member.

Анатомія збою:

  1. Консьюмер вичитав великий батч повідомлень (max.poll.records = 500).
  2. Обробка одного важкого запису зависла (довгий SQL-запит, сторонній HTTP-сервіс).
  3. Час між викликами poll() перевищив ліміт max.poll.interval.ms (дефолт 300 000 мс / 5 хвилин).
  4. Координатор вирішив, що консьюмер завис або помер, виключив його з групи та запустив Rebalance, передавши його партиції іншому консьюмеру.
  5. Коли консьюмер опритомнів і викликав commitSync(), координатор відхилив коміт, оскільки поточне покоління групи (Generation ID) змінилося!

4 Tricky Questions

1. Чому KafkaConsumer.commitAsync() навмисно не виконує повторні спроби (retry) при мережевих збоях, і яка небезпечна гонка даних виникла б, якби він повторював коміти?

Відповідь:

  • Метод commitAsync працює неблокуючим чином.
  • Небезпечний сценарій:
    1. Консьюмер обробив батч 1 і відправив commitAsync(offset = 100).
    2. Запит на офсет 100 завис у мережевому буфері або моргнула мережа.
    3. Консьюмер тут же обробив батч 2 і відправив commitAsync(offset = 200).
    4. Запит з офсетом 200 успішно дійшов до брокера першим, і брокер зафіксував позицію 200.
    5. Якби commitAsync робив retry, то затриманий запит з офсетом 100 повторив би відправлення і прийшов до брокера після офсету 200, перетерши актуальну позицію 200 застарілим значенням 100!
  • У результаті при перезапуску консьюмер перечитав би 100 уже оброблених повідомлень. Тому commitAsync ніколи не робить автоматичних повторів.

2. Що станеться, якщо розробник припуститься помилки та передасть у commitSync() офсет поточного повідомлення record.offset(), а не record.offset() + 1?

Відповідь:

  • Kafka інтерпретує значення committed offset як позицію наступного повідомлення, яке консьюмер хоче отримати при рестарті.
  • Якщо зафіксувати record.offset(), брокер збереже номер поточного повідомлення.
  • При перезапуску сервісу або ребалансі консьюмер запитає читання починаючи зі збереженого офсету.
  • Результат: найостанніше оброблене повідомлення перед перезапуском буде гарантовано прочитане та оброблене повторно (дублікат) при кожному штатному або аварійному рестарті консьюмера.

3. Як оптимізувати застосунок, якщо консьюмер періодично падає з помилкою CommitFailedException?

Відповідь:

  • Для усунення CommitFailedException необхідно виключити затримки між викликами poll(), що перевищують max.poll.interval.ms:
    1. Зменшити розмір пачки: знизити max.poll.records (наприклад, з 500 до 50), щоб консьюмер гарантовано встигав розібрати весь батч за розумний час.
    2. Збільшити допустимий інтервал: підняти max.poll.interval.ms (наприклад, до 10-15 хвилин), якщо важкі фонові операції є очікуваними.
    3. Винести важку обробку у Worker Thread Pool: потік консьюмера займається лише вичитуванням poll() та координацією, а обробку передає пулу потоків CompletableFuture (при цьому потрібне акуратне керування офсетами).
    4. Усунути блокування: додати суворі HTTP-таймаути (ReadTimeout = 2s) при викликах зовнішніх API.

4. Яким чином Kafka справляється з колосальним дисковим навантаженням, коли тисячі сервісів комітять офсети десятки тисяч разів на секунду?

Відповідь:

  • Висока продуктивність комітів забезпечується трьома механізмами:
    1. In-Memory Cache: брокер тримає актуальні офсети в RAM. Читання офсетів ніколи не ходить на диск.
    2. Append-Only Sequential Write: запис у топік __consumer_offsets є суворо послідовним дозаписом у кінець лог-сегмента операційної системи, що дає швидкість дискових операцій, порівнянну із записом у пам’ять.
    3. Log Compaction: фоновий процес cleaner-thread видаляє застарілі записи для однакових ключів [group, topic, partition], зберігаючи лише останній актуальний стан офсету. Розмір топіка на диску залишається компактним і не розростається до нескінченності.

🎯 Шпаргалка для інтерв’ю

Головне про коміт офсетів

  • Сутність: фіксація позиції читання консьюмер-групи в топіку __consumer_offsets.
  • Правило +1: завжди фіксується зміщення наступного повідомлення: record.offset() + 1.
  • Sync vs Async:
    • commitSync: блокуючий, робить retry, надійний, дорогий за затримками.
    • commitAsync: неблокуючий, без retry, швидкий, ідеальний для основного циклу обробки.
  • Золотий стандарт: commitAsync() у тілі циклу poll() і commitSync() у блоці finally перед закриттям.
  • CommitFailedException: виникає, коли час обробки батча перевищує max.poll.interval.ms і консьюмер виключається з групи через ребаланс.
  • Лікування лагів: зменшувати max.poll.records, збільшувати max.poll.interval.ms, оптимізувати важкий I/O в потоках.

Порівняння методів фіксації

| Властивість | commitSync() | commitAsync() | Автокоміт (enable.auto.commit) | | :— | :— | :— | :— | | Блокування потоку | Так (5–50 мс) | Ні (<1 мс) | Ні (фоновий потік/poll) | | Повтор при збої (Retry)| Так (автоматично) | ❌ Ні (захист від гонок) | Ні | | Ризик втрати даних | ❌ Нульовий | Мінімальний | Високий (коміт до обробки) | | Рекомендоване місце | finally перед close() | У кінці ітерації poll() | Лише для локальних тестів |

Червоні прапорці на співбесіді (Чого говорити не можна)

  • ❌ «У commitSync потрібно передавати offset поточного запису, брокер сам додасть одиницю» — брокер нічого не додає, це призведе до повторного читання повідомлення.
  • ❌ «commitAsync завжди робить retry при мережевих помилках» — категорично ні, retry в async комітах заборонено через небезпеку перезапису свіжих офсетів старими.
  • ❌ «Автокоміт раз на 5 секунд — чудовий вибір для мікросервісів» — автокоміт комітить дані до їхньої реальної обробки, що призводить до безповоротної втрати повідомлень при аварії.

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