Что произойдет в случае трех Partition и четырех Consumer в Kafka?

Kafka: 3 Partition и 4 Consumer — что произойдет?

Отличный вопрос про распределение нагрузки в Kafka. Это касается базового механизма consumer groups.

Базовые правила Kafka Consumer Group

Правило 1: Consumer per Partition

Каждая partition может быть обработана ТОЛЬКО одним consumer в группе.
Это гарантирует order сообщений в partition.

Правило 2: Реbalancing

Когда меняется количество consumer, происходит rebalance:
- Консьюмеры временно прекращают обработку
- Kafka переераспределяет partition
- Консьюмеры продолжают

Сценарий: 3 Partition, 4 Consumer

Начальная ситуация

Топик: "orders" с 3 partitions
Consumer group: "payment-service" с 4 consumers

Topic: orders
├── Partition 0
├── Partition 1
└── Partition 2

Consumer group: payment-service
├── Consumer 0 (consumer.id = "payment-0")
├── Consumer 1 (consumer.id = "payment-1")
├── Consumer 2 (consumer.id = "payment-2")
└── Consumer 3 (consumer.id = "payment-3")  ← ЛИШНИЙ!

Что произойдет при rebalance

allocation = {
    "Consumer 0": ["Partition 0"],
    "Consumer 1": ["Partition 1"],
    "Consumer 2": ["Partition 2"],
    "Consumer 3": []  # НЕТ partitions - НИЧЕ НЕ ДЕЛАЕТ
}

# Consumer 3 будет ждать вхолостую (idle)
# Это пустая трата ресурсов

Визуально: процесс Rebalancing

       BEFORE REBALANCE
┌────────────────────────┐
│ Topic: orders          │
│ Partitions: 3          │
└────────────────────────┘
        P0  P1  P2
        ↓   ↓   ↓
        C0  C1  C2  C3
        ↓   ↓   ↓   ↓
    [working] [idle]

       DURING REBALANCE (stop the world)
    [all consumers paused]

       AFTER REBALANCE
        P0  P1  P2
        ↓   ↓   ↓
        C0  C1  C2  C3
    [working...]  [idle]

Код: как это выглядит

from kafka import KafkaConsumer
from kafka import KafkaProducer
import json

# PRODUCER - добавляет сообщения в partitions
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

for i in range(100):
    message = {'order_id': i, 'amount': 100.00}
    # Сообщение идёт в одну из 3 partitions
    producer.send('orders', value=message)

# CONSUMERS - группа из 4 консьюмеров
def create_consumer(consumer_id):
    consumer = KafkaConsumer(
        'orders',
        group_id='payment-service',
        bootstrap_servers=['localhost:9092'],
        value_deserializer=lambda m: json.loads(m.decode('utf-8')),
        auto_offset_reset='earliest'
    )
    return consumer

# Запустим 4 consumer'а
consumers = []
for i in range(4):
    c = create_consumer(f'consumer-{i}')
    consumers.append(c)

# REBALANCING происходит автоматически
# Kafka reassigns partitions

# Результат:
# Consumer 0: читает Partition 0
# Consumer 1: читает Partition 1
# Consumer 2: читает Partition 2
# Consumer 3: IDLE (ничего не делает)

Что происходит внутри

Процесс Rebalancing

# 1. Trigger (одна из этих причин)
trigger = {
    "new_consumer_joins": "Запустили 4-го consumer",
    "consumer_crashes": "Один consumer упал",
    "timeout": "Consumer не отправлял heartbeat > session.timeout.ms",
    "manual": "КОД вызвал leave_group()"
}

# 2. Stop the world
# Все consumers прекращают обработку
# В это время: потеря throughput

# 3. Group coordinator выполняет rebalance
# Обычно: RoundRobin или Range strategy

strategy = {
    "RoundRobin": {
        "logic": "Распределяет consumers по partitions по кругу",
        "result_3p_4c": {
            "C0": ["P0"],
            "C1": ["P1"],
            "C2": ["P2"],
            "C3": []  # idle
        }
    },
    "Range": {
        "logic": "Распределяет диапазоны partitions",
        "result_3p_4c": {
            "C0": ["P0"],
            "C1": ["P1"],
            "C2": ["P2"],
            "C3": []  # idle
        }
    }
}

# 4. Все consumers вновь начинают читать
# Consumer 3: остаётся без работы

Жизненный цикл Rebalancing

import time
from datetime import datetime

print("[00:00] Consumer 3 запущен")
print("[00:01] REBALANCE START")
print("        - Все consumers перестают читать")
print("        - Group coordinator переопределяет assignment")
print("        - Каждый consumer коммитит offset'ы")
print("[00:03] REBALANCE COMPLETE")
print("        - C0: читает P0 (с offset 150)")
print("        - C1: читает P1 (с offset 200)")
print("        - C2: читает P2 (с offset 180)")
print("        - C3: IDLE (нет assignment)")
print("[00:05] Потеря данных: нет")
print("        Потеря throughput: ~3-5 сек (время rebalance)")
print("        Потеря ресурсов: Consumer 3 работает вхолостую")

Метрики Rebalancing

# Как мониторить rebalancing

metrics = {
    "rebalance_duration": "Сколько времени занял rebalance (на 1-2 сек может быть нормально)",
    "rebalance_latency": "Как долго messages не обрабатывались",
    "lag_spikes": "Потребитель отставает из-за rebalance",
    "assignment_balance": "Хорошо ли распределены partitions"
}

# Логи (при enable debug logging)
rebalance_logs = """
[GroupCoordinator id=1] Executing onJoinLeader for group payment-service, 
  generation 42, membersCount 4, joinedCount 3
[GroupCoordinator id=1] Rebalancing group payment-service into 4 generations:
  payment-service-0 -> [Partition 0]
  payment-service-1 -> [Partition 1]
  payment-service-2 -> [Partition 2]
  payment-service-3 -> []
"""

Что не произойдет (важные моменты)

what_wont_happen = {
    "data_loss": {
        "concern": "Потеряются ли сообщения?",
        "answer": "НЕТ. Все сообщения остаются в partitions"
    },
    "message_duplication": {
        "concern": "Сообщения обработаны дважды?",
        "answer": "Возможно, если не коммитить offset'ы правильно",
        "solution": "Используй exactly-once semantics (transactional reads)"
    },
    "order_guarantee": {
        "concern": "Порядок сообщений сохранится?",
        "answer": "ДА. В каждой partition порядок гарантирован"
    },
    "stuck_messages": {
        "concern": "Застрянут ли сообщения?",
        "answer": "НЕТ. Consumer 3 просто idle, сообщения не читаются, но не теряются"
    }
}

Оптимизация: как избежать idle consumer

Вариант 1: Увеличить количество partitions

# Было: 3 partitions, 4 consumers
# Сейчас: каждый consumer получит свою partition (если еще добавить)

# Решение:
# 1. Увеличить partitions: 3 → 4
# 2. Переопределить существующие данные (если нужно)

config = {
    "topic": "orders",
    "partitions": 4,  # было 3
    "replication_factor": 3
}

# Теперь:
# Consumer 0 -> Partition 0
# Consumer 1 -> Partition 1
# Consumer 2 -> Partition 2
# Consumer 3 -> Partition 3  (ура! теперь работает)

Вариант 2: Уменьшить количество consumers

# Было: 3 partitions, 4 consumers
# Решение: запустить только 3 consumer'а

for i in range(3):  # было range(4)
    c = create_consumer(f'consumer-{i}')
    consumers.append(c)

# Теперь каждый работает на полную

Вариант 3: Несколько consumer groups

# Разные group_id -> разные assignments

# Payment processing
payment_consumer = KafkaConsumer(
    'orders',
    group_id='payment-service',  # Group 1
    bootstrap_servers=['localhost:9092']
)

# Analytics
analytics_consumer = KafkaConsumer(
    'orders',
    group_id='analytics-service',  # Group 2 (другой!)
    bootstrap_servers=['localhost:9092']
)

# Каждая группа независимо читает все messages
# Разные offset'ы для каждой группы

Лучшие практики

best_practices = {
    "rule_1": "Количество consumers ≤ Количество partitions",
    "reason": "Excess consumers будут idle",
    "example": "3 partitions -> max 3 useful consumers",
    
    "rule_2": "Плануй scaling: 1 partition = 1 consumer",
    "reason": "Каждый consumer обрабатывает 1 partition (максимум)",
    "math": "Throughput = (messages per partition) / (processing time per consumer)",
    
    "rule_3": "Мониторь rebalancing time",
    "reason": "Долгий rebalance = потеря throughput",
    "target": "< 5 сек для 100 consumers",
    
    "rule_4": "Используй static membership (Kafka 2.3+)",
    "reason": "Избежать unnecessary rebalance при restart",
    "code": "consumer_config['group.instance.id'] = 'consumer-1'"
}

Сценарии в production

Сценарий 1: Scaling up

Текущее: 3 partitions, 3 consumers (каждый обрабатывает 1 partition)
Проблема: Throughput упал, нужна скорость

Решение:
1. Увеличить partitions: 3 → 6
2. Запустить еще 3 consumers
3. Происходит rebalance (~2 сек downtime)
4. Теперь: 6 partitions, 6 consumers
   Throughput удвоился

Сценарий 2: Consumer crash

Текущее: Consumer 2 упал (был на Partition 2)
Каф видит: heartbeat timeout (default 10 сек)
Каф делает: rebalance

Новое распределение:
- Consumer 0: Partition 0 (как было)
- Consumer 1: Partition 1 (как было)
- Consumer 2: DOWN
- Partition 2: Переходит к Consumer 0 или 1

Результат: Partition 2 обрабатывается, но медленнее

Итог

В сценарии 3 Partition, 4 Consumer:

result = {
    "assignment": "C0:P0, C1:P1, C2:P2, C3:IDLE",
    "data_loss": False,
    "processing": "Продолжается нормально (3 consumers работают)",
    "rebalance_time": "~1-3 сек (stop the world)",
    "problem": "Consumer 4 пустует, вкладываем ресурсы впустую",
    "solution": "Либо +1 partition, либо -1 consumer, либо разные group_ids"
}

Золотое правило:

ideal_consumers = partitions  # или partitions + резерв (N+1)