Что произойдет, если появится еще один Consumer, при неизменном количестве Partition

Kafka: добавление Consumer при фиксированном количестве Partitions

Это вопрос про распределённую систему обработки сообщений Apache Kafka. Добавление нового Consumer в Consumer Group с неизменным количеством Partitions приводит к rebalancing и перераспределению нагрузки.

1. Исходная ситуация

Тема (Topic): orders
  Partitions: 3  (Partition-0, Partition-1, Partition-2)
  
Consumer Group: payment-service
  Consumers: 2 (Consumer-1, Consumer-2)
  
ДО добавления:
  Consumer-1 → Partition-0, Partition-1 (обрабатывает 2 партиции)
  Consumer-2 → Partition-2 (обрабатывает 1 партицию)

2. Что происходит при добавлении третьего Consumer

ШАГ 1: Новый Consumer присоединяется к группе
  Consumer-3 → присоединился
  
ШАГ 2: Kafka запускает REBALANCING
  - Все текущие consumers приостанавливают работу
  - Kafka вычисляет новое распределение
  - Каждый consumer получает список своих partitions
  
ШАГ 3: Новое распределение (Round-Robin по умолчанию)
  Consumer-1 → Partition-0
  Consumer-2 → Partition-1
  Consumer-3 → Partition-2
  
ШАГ 4: Consumers возобновляют работу
  - Каждый consumer начинает читать из своих partitions
  - Обрабатывают очереди сообщений

3. Проблемы при rebalancing

// Во время rebalancing происходит:
1. STOP - все consumers останавливаются (обработка прерывается)
2. REBALANCE - пересчёт распределения (может быть долго)
3. OFFSET - каждый consumer восстанавливает позицию чтения
4. RESUME - продолжение обработки

// Временная линия:
Время 0ms:    Consumer-3 присоединяется
Время 0-50ms: Consumers обнаруживают изменение
Время 50-200ms: Rebalancing происходит
Время 200ms:  Новое распределение активно

// ПОТЕРЯ ПРОИЗВОДИТЕЛЬНОСТИ: 200ms без обработки!

4. Сценарий с Kafka в Java

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class PaymentConsumer {
    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList("orders"));
        
        // Consumer-1 работает
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            // Обработка сообщений
            for (ConsumerRecord<String, String> record : records) {
                processOrder(record);
            }
        }
    }
    
    // Когда Consumer-2 присоединяется:
    // 1. consumer.poll() вернёт ПУСТОЙ список (rebalancing)
    // 2. Произойдёт переназначение partitions
    // 3. Consumer-1 теперь обрабатывает только Partition-0
    // 4. Работа возобновляется
}

5. Сравнение сценариев

Сценарий 1: Partitions > Consumers
  3 Partitions, 2 Consumers → добавляем 3-го Consumer
  Распределение: каждый Consumer получает ровно 1 Partition
  Результат: ЛУЧШЕ (идеальное распределение)
  
Сценарий 2: Partitions = Consumers (ТЕКУЩИЙ ВОПРОС)
  3 Partitions, 2 Consumers → добавляем 3-го Consumer
  Было: Consumer-1 обрабатывает 2, Consumer-2 обрабатывает 1
  Стало: каждый обрабатывает 1
  Результат: ХОРОШО (более равномерно)
  
Сценарий 3: Partitions < Consumers
  2 Partitions, 2 Consumers → добавляем 3-го Consumer
  Consumer-3 будет idle (ничего не будет обрабатывать!)
  Результат: ПЛОХО (зря запущен Consumer)

6. Как работает Assignment Strategy

// RoundRobinAssignor (по умолчанию часто)
Partitions: [0, 1, 2]
Consumers: [Consumer-1, Consumer-2, Consumer-3]

Assignment:
Consumer-1 ← Partition-0
Consumer-2 ← Partition-1
Consumer-3 ← Partition-2

// RangeAssignor
Consumer-1 ← Partition-0
Consumer-2 ← Partition-1  
Consumer-3 ← Partition-2

// StickyAssignor (минимизирует пересаживание)
// Старое: Consumer-1 [0, 1], Consumer-2 [2]
// Новое: Consumer-1 [0], Consumer-2 [1], Consumer-3 [2]
// Изменилось минимально (лучше для performance)

7. Дополнительные параметры при rebalancing

Properties props = new Properties();

// session.timeout.ms - сколько времени ждать ответ от consumer
// Default: 10000ms (10 сек)
props.put("session.timeout.ms", "10000");

// heartbeat.interval.ms - как часто отправлять heartbeat
// Default: 3000ms (3 сек)
props.put("heartbeat.interval.ms", "3000");

// max.poll.interval.ms - максимум между poll() вызовами
// Default: 300000ms (5 минут)
props.put("max.poll.interval.ms", "300000");

// Если Consumer-2 зависнет > session.timeout.ms
// → Kafka считает его мёртвым
// → Запускает rebalancing

8. Мониторинг при добавлении Consumer

// Слушать события rebalancing
public class RebalanceListener implements ConsumerRebalanceListener {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Rebalancing: отзываю partitions " + partitions);
        // Сохранить offset перед потерей
    }
    
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        System.out.println("Rebalancing: мне даны partitions " + partitions);
        // Подготовиться к обработке новых partitions
    }
}

consumer.subscribe(Arrays.asList("orders"), new RebalanceListener());

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

✅ ХОРОШО:
1. Количество Partitions ≥ Количество Consumers
   (каждый Consumer обрабатывает хотя бы 1 Partition)

2. Добавляй Consumers постепенно
   (избегай резких скачков нагрузки)

3. Используй StickyAssignor
   (минимизирует perebalancing overhead)

4. Мониторь rebalancing time
   (если > 5 сек - проблема)

5. Сохраняй offsets в БД
   (на случай если Consumer упадёт)

❌ ПЛОХО:
1. Добавлять Consumers > Partitions
   (некоторые будут idle)

2. Часто добавлять/удалять Consumers
   (постоянный rebalancing)

3. Обрабатывать долгие операции в poll()
   (может вызвать session timeout)

4. Не обрабатывать исключения при rebalancing
   (потеря данных)

10. Итоговая схема

ДО:    Consumer-1 ━━ Partition-0
       Consumer-1 ━━ Partition-1
       Consumer-2 ━━ Partition-2

↓ ДОБАВЛЯЕМ Consumer-3 + REBALANCING ↓

ПОСЛЕ:  Consumer-1 ━━ Partition-0
        Consumer-2 ━━ Partition-1
        Consumer-3 ━━ Partition-2
        
РЕЗУЛЬТАТ:
- Нагрузка распределена равномерно
- Пропускная способность увеличена в 1.5 раза (было 2-1, стало 1-1-1)
- Во время rebalancing: downtime 50-200ms

Вывод

При добавлении нового Consumer к Consumer Group с неизменным количеством Partitions происходит rebalancing: все Consumers останавливаются, Kafka пересчитывает распределение, каждый Consumer получает новый набор Partitions. Результат: более равномерное распределение нагрузки, но временный downtime на время rebalancing (обычно 50-200ms). Это нормальный и ожидаемый процесс в Kafka при масштабировании.