📨 Раздел 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 секунд — отличный выбор для микросервисов» — автокоммит коммитит данные до их реальной обработки, что приводит к безвозвратной потере сообщений при аварии.

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