JavaNotes

Kafka Consumer

3 min read

Kafka Consumer

Consumer получает records из partition через leader broker. Порядок гарантирован только внутри одной partition: между разными partitions общего порядка нет.

Consumer group

Внутри одной group каждая partition назначена не более чем одному consumer одновременно. Один consumer может читать несколько partitions; лишние consumers относительно числа partitions простаивают.

topic: orders
P0 -> consumer-A
P1 -> consumer-B
P2 -> consumer-A

При входе/выходе consumer, изменении partitions или проблеме с heartbeat происходит rebalance. Во время переназначения обработка может приостанавливаться, поэтому долгую работу внутри poll loop согласуют с max.poll.interval.ms и выбранной стратегией обработки.

Разные groups читают один topic независимо и имеют собственные offsets.

Position и committed offset

  • position — следующая запись, которую текущий consumer собирается получить;
  • committed offset — сохранённая для group позиция восстановления после restart/rebalance.

Коммитят offset следующей записи, а не последней обработанной. После успешной обработки record с offset 41 сохраняют 42.

Offsets хранятся Kafka во внутреннем compacted topic. Auto-commit упрощает код, но момент коммита может не совпасть с фактическим завершением бизнес-операции.

At-most-once

  1. Сохранить offset.
  2. Обработать record.

Если процесс упадёт между шагами, record потеряется для этой group. Повтор не произойдёт, но возможна потеря — не более одного раза.

Такой режим допустим для данных, потеря которых несущественна, но редко подходит для бизнес-событий.

At-least-once

  1. Получить record.
  2. Успешно обработать.
  3. Сохранить offset.

Если процесс упадёт после побочного эффекта, но до commit offset, record придёт повторно. Потери из-за этого окна нет, но возможен дубль — как минимум один раз.

Обработчик делают идемпотентным:

  • уникальный event_id и таблица обработанных событий;
  • upsert по бизнес-ключу;
  • проверка версии агрегата;
  • транзакционное сохранение результата и inbox-записи в одной БД.

Ручной commit

while (running) {
    ConsumerRecords<String, Event> records =
            consumer.poll(Duration.ofMillis(500));
 
    for (ConsumerRecord<String, Event> record : records) {
        handler.handle(record.value());
    }
 
    consumer.commitSync();
}

Это at-least-once только если handle действительно завершил все необходимые эффекты до commit. При ошибке посередине batch часть уже обработанных records может повториться.

commitAsync не блокирует poll, но callbacks старых и новых commits способны завершиться в другом порядке. На shutdown/rebalance часто нужен контролируемый sync commit подтверждённой позиции.

Ошибки и poison record

Бесконечный retry одной некорректной записи останавливает partition. Политика должна различать:

  • временную ошибку — ограниченный retry с backoff;
  • постоянную ошибку данных — отдельный quarantine/DLT поток;
  • ошибку кода — сигнал и исправление, а не бесконечное скрытие.

Перед отправкой в DLT сохраняют исходные topic, partition, offset, key, причину и версию обработчика, не раскрывая секреты.

Exactly-once

Kafka transactions позволяют атомарно записать результаты в Kafka и offsets прочитанной group в одну transaction. Потребитель результата должен читать с isolation.level=read_committed.

Это даёт exactly-once semantics для поддерживаемого consume-transform-produce пути внутри Kafka. HTTP-вызов, письмо или запись в независимую БД не становятся атомарными автоматически.

acks и надёжность producer

acks — настройка producer, а не consumer:

  • acks=0 не ждёт подтверждения;
  • acks=1 ждёт leader;
  • acks=all ждёт подтверждения согласно in-sync replicas.

acks=all работает вместе с replication factor и min.insync.replicas. Он снижает риск потери записи, но не определяет, сколько раз её обработает consumer.

На собеседовании

Почему at-least-once создаёт дубли? Сбой возможен после эффекта, но до commit offset; после восстановления record читается снова.

Что произойдёт, если consumers больше partitions? Часть consumers group не получит partition и будет простаивать.

Где гарантирован порядок? Только внутри partition. Key обычно используют, чтобы связанные события попадали в одну partition.

Связанные заметки: Kafka, Kafka Broker, Kafka Producer, Команды Kafka.

Graph View

Backlinks