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

Как работает балансировка консьюмеров в группе

Балансировка — это распределение авиарейсов между авиадиспетчерами:

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

🟢 Junior Level

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

Балансировка консьюмеров (Partition Assignment) — это автоматический распределённый механизм Apache Kafka, который динамически распределяет партиции одного или нескольких топиков между активными консьюмерами внутри одной Consumer Group.

Зачем она нужна:

  1. Автоматизация: разработчику не нужно вручную прописывать, какой сервер какую партицию читает.
  2. Эластичность: при падении или добавлении новых инстансов микросервиса брокер инициирует ребалансировку (Rebalance) и справедливо перераспределяет партиции между оставшимися живыми участниками.
  3. Строгий инвариант: внутри группы одна партиция назначается строго одному консьюмеру. При этом один консьюмер может читать сразу несколько партиций.

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

Балансировка — это распределение авиарейсов между авиадиспетчерами:

  • В аэропорту 6 посадочных полос (6 партиций топика).
  • На смену заступили 3 авиадиспетчера (3 консьюмера в группе).
  • Старший диспетчер выделяет каждому ровно по 2 полосы: Диспетчер А следит за полосами 0 и 1, Диспетчер Б — за 2 и 3, Диспетчер В — за 4 и 5.
  • Если Диспетчер В внезапно теряет сознание (краш сервиса), старший диспетчер немедленно объявляет перераспределение и передает полосы 4 и 5 оставшимся диспетчерам А и Б. Ни одна полоса не остаётся без контроля!

Базовый пример структуры распределения

[ Топик: 6 партиций ]   P0    P1    P2    P3    P4    P5
                         │     │     │     │     │     │
                         └──┬──┘     └──┬──┘     └──┬──┘
                            ▼           ▼           ▼
                      [Consumer A] [Consumer B] [Consumer C]
                      (2 партиции) (2 партиции) (2 партиции)

В коде Java консьюмер просто указывает group.id и подписывается на топик:

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "orders-processing-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders")); // Kafka сама распределит партиции

🟡 Middle Level

4 основные стратегии распределения (PartitionAssignor)

Выбор стратегии задаётся параметром partition.assignment.strategy:

Стратегия Принцип работы Плюсы Минусы
RangeAssignor (исторический дефолт) Рассматривает каждый топик изолированно и делит партиции на непрерывные диапазоны. Простая реализация. Колоссальный перекос нагрузки при подписке на несколько топиков!
RoundRobinAssignor Собирает все партиции всех топиков в один список и раздает по кругу консьюмерам. Равномерное число партиций у всех участников. Полностью перемешивает партиции при любом ребалансе.
StickyAssignor Сохраняет максимальное количество текущих назначений партиций при ребалансе. Минимизирует миграцию партиций, бережет кэш данных. Использует устаревший Eager протокол (Stop-the-World).
CooperativeStickyAssignor (современный стандарт) Инкрементальная ребалансировка без остановки нетронутых партиций. 🚀 Минимальное время простоя, отсутствие глобального Stop-the-World. Доступен с Kafka 2.3+.

Проблема перекоса RangeAssignor (Range Skew)

Представьте, что микросервис подписан на 10 топиков, и в каждом топике по 3 партиции (P0, P1, P2). Всего в группе 2 консьюмера (C1 и C2):

  • Формула RangeAssignor для каждого топика: $3 \div 2 = 1.5 \to$ Консьюмер C1 получает 2 партиции (P0, P1), а Консьюмер C2 получает 1 партицию (P2).
  • Этот расчет повторяется для каждого из 10 топиков!
  • В итоге: Консьюмер C1 получает 20 партиций, а Консьюмер C2 — всего 10 партиций!
  • Консьюмер C1 перегружен в 2 раза сильнее, упирается в CPU, накапливает лаг, а C2 недогружен.
  • Вывод: никогда не используйте RangeAssignor, если группа читает несколько топиков!

Роль ConsumerRebalanceListener

При возникновении ребалансировки консьюмер должен успеть корректно сохранить свое состояние:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Вызывается ДО того, как брокер отберет партиции.
        // ОБЯЗАТЕЛЬНО: синхронно коммитим смещения, чтобы другой узел не обработал дубли!
        consumer.commitSync();
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // Вызывается ПОСЛЕ того, как брокер назначил консьюмеру новые партиции.
        // Инициализируем локальные кэши, смещения в БД или метрики.
    }
});

🔴 Senior Level

Анатомия классического протокола ребалансировки (JoinGroup / SyncGroup)

Многие ошибочно думают, что сам брокер Kafka распределяет партиции между клиентами. Это не так! В классическом протоколе Kafka распределение выполняет один из клиентских консьюмеров (Leader):

[ Consumer A ]               [ Group Coordinator (Брокер) ]               [ Consumer B ]
      │                                    │                                    │
      │ 1. JoinGroup Request               │                                    │
      ├───────────────────────────────────►│ 1. JoinGroup Request               │
      │                                    │◄───────────────────────────────────┤
      │                                    │                                    │
      │ (Брокер выбирает Лидером Client A) │                                    │
      │                                    │                                    │
      │ 2. JoinGroup Response              │ 2. JoinGroup Response              │
      │ (allMembers + YOU ARE LEADER)      │ (memberId + LEADER IS A)           │
      │◄───────────────────────────────────┼───────────────────────────────────►│
      │                                    │                                    │
      │ [ Лидер A локально выполняет       │                                    │
      │   PartitionAssignor.assign() ]     │                                    │
      │                                    │                                    │
      │ 3. SyncGroup Request               │ 3. SyncGroup Request               │
      │ (отправляет готовый план для всех) │ (пустой запрос ожидания)           │
      ├───────────────────────────────────►│◄───────────────────────────────────┤
      │                                    │                                    │
      │ 4. SyncGroup Response              │ 4. SyncGroup Response              │
      │ (получает свои партиции: P0, P1)   │ (получает свои партиции: P2)       │
      │◄───────────────────────────────────┼───────────────────────────────────►│

KIP-848: Революция протокола ребалансировки (Kafka 3.7+ / 4.0)

Классический протокол (KIP-429 и ниже) имел архитектурный недостаток: зависимость от клиента-лидера и двухфазный опрос JoinGroup $\to$ SyncGroup. В рамках KIP-848 (Next Generation Consumer Protocol):

  1. Server-Side Assignment: логика вычисления распределения перенесена полностью на сторону Group Coordinator (Брокера).
  2. Heartbeat-Driven Reconciliation: исчезли блокирующие фазы JoinGroup/SyncGroup. Консьюмер просто отправляет регулярный RPC-запрос ConsumerGroupHeartbeat.
  3. Брокер в теле ответа Heartbeat инкрементально возвращает директивы: «Отдай партицию 2, возьми партицию 5». Ребаланс стал полностью бесшовным и прозрачным.

Катастрофическая петля HPA (Kubernetes Autoscaling Death Loop)

Распространённый антипаттерн в архитектуре облачных микросервисов:

  1. DevOps настраивают Kubernetes Horizontal Pod Autoscaler (HPA) на основе утилизации процессора (targetCPUUtilizationPercentage: 70%).
  2. В топик поступает всплеск сообщений. Нагрузка на поды консьюмеров вырастает до 85% CPU.
  3. HPA видит перегрузку и поднимает 3 новых пода.
  4. Новые поды вступают в Consumer Group, что инициирует ребалансировку.
  5. При ребалансировке потоки останавливаются, коммитят оффсеты, переинициализируют соединения. Сетевые стеки и GC кратковременно загружают процессор на 100%.
  6. HPA фиксирует рост нагрузки до 95% CPU и поднимает еще 5 подов!
  7. Начинается непрерывный шторм ребалансировок, обработка сообщений полностью блокируется, лаг улетает в миллионы записей.
    • Решение: масштабировать Kafka-консьюмеры в Kubernetes только по бизнес-метрике Consumer Lag (через KEDA или Prometheus Adapter) и строго с защитным интервалом охлаждения (cooldown period не менее 5–10 минут).

4 Tricky Questions

1. Почему RangeAssignor приводит к катастрофическому перекосу нагрузки при вычитке нескольких топиков, и почему его до сих пор не удалили?

Ответ:

  • RangeAssignor группирует партиции каждого топика строго по числовому порядку и делит их на количество консьюмеров независимыми диапазонами.
  • Если у вас есть 5 топиков по 3 партиции (0, 1, 2) и 2 консьюмера (C1 и C2), деление $3 / 2 = 1.5$ округляется в пользу первого консьюмера. В каждом из 5 топиков консьюмер C1 получает партиции 0 и 1, а C2 — партицию 2.
  • В сумме C1 будет читать 10 партиций, а C2 — всего 5 партиций (в 2 раза меньше!). Если топиков будет 50, C1 получит 100 партиций, а C2 — 50.
  • Причиной, почему он оставался дефолтным долгие годы, была обратная совместимость (Colocated Partitions): если продюсер пишет связанные данные в топики A и B по одному ключу в одинаковые номера партиций, RangeAssignor гарантировал, что партиция 0 топика A и партиция 0 топика B попадут на один и тот же инстанс консьюмера (удобно для локальных JOIN в ранних версиях).

2. Кто физически исполняет Java-код стратегии распределения партиций (PartitionAssignor.assign()) — брокер Kafka или клиент?

Ответ:

  • В классическом протоколе Kafka (до KIP-848) алгоритм assign() исполняется на одном из клиентских инстансов консьюмеров, выбранном в качестве Лидера группы (Group Leader).
  • Брокер (Group Coordinator) собирает список всех участников группы через запросы JoinGroup и передает этот список первому подключившемуся консьюмеру (Лидеру).
  • Лидер локально в своем JVM-процессе запускает Java-код настроенного класса PartitionAssignor, формирует Map<MemberId, List<TopicPartition>> и отправляет этот результат обратно брокеру в запросе SyncGroup.
  • И только начиная с KIP-848 (Kafka 3.7+ / 4.0) брокер берет расчет назначений на себя (Server-Side Assignor).

3. Что произойдет, если в методе ConsumerRebalanceListener.onPartitionsRevoked() произойдет необработанный RuntimeException (например, таймаут БД)?

Ответ:

  • Если метод onPartitionsRevoked() выбрасывает исключение, консьюмер прерывает фазу подготовки к ребалансу.
  • Несохраненные смещения обработанных записей не будут зафиксированы в __consumer_offsets.
  • Сам инстанс консьюмера признает внутреннее состояние несогласованным, прерывает цикл poll() и закрывает клиентскую сессию с выбросом критической ошибки KafkaException.
  • Партиции отзываются брокером принудительно по таймауту. Новый консьюмер, которому брокер переназначит эти партиции, начнет вычитку со старых закоммиченных оффсетов, что неминуемо приведет к повторной обработке пачки сообщений (Duplicate Processing).

4. Чем принципиально отличается поведение StickyAssignor от CooperativeStickyAssignor при добавлении нового инстанса в работающую группу?

Ответ:

  • StickyAssignor использует протокол Eager: когда новый участник входит в группу, брокер требует от ВСЕХ работающих участников немедленно прекратить работу и сдать ВСЕ свои партиции. Вся группа встает на паузу (Stop-the-World). Затем StickyAssignor назначает большинству подов их старые партиции, а часть партиций отдает новичку. Пауза затрагивает 100% участников.
  • CooperativeStickyAssignor использует протокол Cooperative (KIP-429): при добавлении новичка работающие консьюмеры НЕ останавливают вычитку своих партиций. На первом шаге брокер вычисляет, какие именно 2–3 партиции нужно забрать для новичка, и просит отдать только их. Все остальные 20 консьюмеров продолжают непрерывно обрабатывать трафик без малейшей паузы!

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

Главное о балансировке

  • Суть: автоматическое распределение партиций топика между консьюмерами группы.
  • Инвариант: 1 партиция = строго 1 консьюмер группы в любой момент времени.
  • Стратегия для Production: CooperativeStickyAssignor (минимальный простой, инкрементальный перенос).
  • Опасность RangeAssignor: катастрофический перекос партиций при подписке на несколько топиков.
  • Кто считает назначения: в классическом протоколе — клиентский консьюмер-Лидер группы (не брокер!). В KIP-848 — брокер.
  • ConsumerRebalanceListener: обязателен для синхронного коммита оффсетов в onPartitionsRevoked().
  • HPA в Kubernetes: масштабировать поды консьюмеров по CPU запрещено (вызывает шторм ребалансировок); масштабировать только по Consumer Lag с cooldown period.

Сравнение стратегий PartitionAssignor

| Стратегия | Протокол | Равномерность | Перемещение партиций | | :— | :— | :— | :— | | RangeAssignor | Eager (Stop-the-World) | ❌ Очень плохая на multi-topic | Среднее | | RoundRobinAssignor | Eager (Stop-the-World) | ✅ Равномерная | ❌ Перемешивает всё подряд | | StickyAssignor | Eager (Stop-the-World) | ✅ Равномерная | ✅ Минимальное перемещение | | CooperativeStickyAssignor | Cooperative (Non-blocking) | ✅ Равномерная | 🚀 Инкрементальное без пауз |

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

  • ❌ «Брокер Kafka сам решает в Java-коде, кому какую партицию отдать» — в стандартной Kafka распределение считает клиент-Лидер группы.
  • ❌ «RangeAssignor всегда делит партиции поровну» — при нечетных партициях на нескольких топиках перекос достигает сотен процентов.
  • ❌ «При ребалансировке вся система всегда останавливается на 30 секунд» — с CooperativeStickyAssignor остановка происходит инкрементально за миллисекунды.

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