Consumer Groups

Consumer groups — это механизм горизонтального масштабирования потребления данных из Kafka. Идея проста: несколько consumer объединяются в группу с общим group.id и совместно читают партиции топика — каждая партиция назначается ровно одному consumer в группе. Добавляете consumer — нагрузка распределяется. Убираете consumer — его партиции переходят к остальным.

group.id: логическое имя группы

group.id — это строковой идентификатор, объединяющий consumer в одну группу. Все consumer с одним group.id кооперируют друг с другом через group coordinator — специальный брокер, ответственный за управление группой.

Назначение партиций

Назначение партиций происходит через алгоритм partition assignment strategy. Kafka поддерживает несколько стратегий:

• RangeAssignor (по умолчанию в legacy protocol): последовательные партиции назначаются consumer по диапазонам
• RoundRobinAssignor: партиции распределяются по очереди, более равномерно при нескольких топиках
• StickyAssignor: минимизирует перемещение партиций при ребалансировке
• CooperativeStickyAssignor: cooperative incremental ребалансировка с минимальным прерыванием
• Server-side assignment (KIP-848): брокер сам управляет назначением — consumer только сообщают о желаемых топиках

ConsumerGroupDiagram: визуализация группы

Диаграмма выше показывает типичное распределение для группы из 2 consumer и 4 партиций: Consumer 0 получает P0 и P1, Consumer 1 получает P2 и P3. Group Coordinator хранит offsets в __consumer_offsets.

Group Coordinator: выбор и роль

Group Coordinator — это брокер, ответственный за жизненный цикл конкретной consumer group. Он определяется детерминированно:

coordinator_partition = hash(group.id) % num_partitions(__consumer_offsets)
coordinator_broker = leader(partition coordinator_partition of __consumer_offsets)

По умолчанию __consumer_offsets имеет 50 партиций. Для group.id='analytics': если hash('analytics') % 50 = 12, то coordinator — это брокер, который является лидером партиции 12 топика __consumer_offsets.

Обязанности Group Coordinator:

Масштабирование: consumer vs партиции

Максимальный параллелизм consumer group ограничен числом партиций топика. Это фундаментальное ограничение архитектуры.

# Правило: добавляем consumer до num_partitions — дальше бессмысленно
# Увеличить параллелизм можно только увеличив число партиций топика

from kafka import KafkaConsumer

# Consumer 1 — в группе 'payments-group'
consumer1 = KafkaConsumer(
    bootstrap_servers=['localhost:9092'],
    group_id='payments-group',
    auto_offset_reset='earliest',
)
consumer1.subscribe(['payments'])

# Consumer 2 — в той же группе 'payments-group'
consumer2 = KafkaConsumer(
    bootstrap_servers=['localhost:9092'],
    group_id='payments-group',
    auto_offset_reset='earliest',
)
consumer2.subscribe(['payments'])
# consumer1 и consumer2 поделят партиции топика 'payments' между собой

Несколько consumer groups: независимое чтение

Различные consumer groups читают один топик полностью независимо. Каждая группа ведёт свои offsets. Это ключевое отличие Kafka от традиционных message broker.

# Три разные группы — три независимых потребителя одного топика
billing_consumer = KafkaConsumer(
    bootstrap_servers=['localhost:9092'],
    group_id='billing-group',   # отдельный offset
)
analytics_consumer = KafkaConsumer(
    bootstrap_servers=['localhost:9092'],
    group_id='analytics-group', # свой независимый offset
)
# Обе группы читают топик 'orders' — полностью независимо
billing_consumer.subscribe(['orders'])
analytics_consumer.subscribe(['orders'])

Мониторинг consumer group lag

Consumer lag — это разница между последним записанным offset (Log End Offset, LEO) и последним закоммиченным offset группы. Лаг показывает, насколько группа отстаёт от продюсеров.

# CLI: мониторинг лага consumer group (Kafka 4.0)
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group analytics-group

# Пример вывода:
# GROUP           TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# analytics-group orders  0          5000            5200            200
# analytics-group orders  1          4800            5100            300
# analytics-group orders  2          5100            5100            0

Суммарный лаг 200 + 300 + 0 = 500 означает, что группа не обработала 500 сообщений. Если лаг растёт — producer пишет быстрее, чем consumer обрабатывает. Решение: добавить consumer (до числа партиций) или увеличить число партиций.

Итог

Consumer groups — фундаментальный механизм масштабирования потребления в Kafka. group.id объединяет consumer. Group coordinator управляет назначением партиций и offsets. Максимальный параллелизм = число партиций. Несколько groups читают независимо. Лаг — ключевая метрика здоровья группы.