Почему для каждого Consumer должна быть отдельная партиция?

Почему для каждого Consumer должна быть отдельная партиция в Kafka?

На самом деле, это утверждение не совсем точно. Система Kafka не требует, чтобы у каждого consumer-а была отдельная партиция. Однако это рекомендуемая архитектурная практика для оптимальной производительности. Разберёмся в деталях.

Основной принцип распределения партиций в Kafka

Партиция — это неделимая единица параллелизма в Kafka. Каждая партиция может быть прочитана только одним consumer-ом в пределах одной consumer group в один момент времени.

Топик: payments
├── Партиция 0 → Consumer-1 (Consumer Group A)
├── Партиция 1 → Consumer-2 (Consumer Group A)
├── Партиция 2 → Consumer-3 (Consumer Group A)
└── Партиция 3 → Consumer-4 (Consumer Group A)

Или:

Топик: payments
├── Партиция 0 → Consumer-1 (Consumer Group B)
├── Партиция 1 → Consumer-1 (Consumer Group B)
├── Партиция 2 → Consumer-2 (Consumer Group B)
└── Партиция 3 → Consumer-2 (Consumer Group B)

Идеальное распределение: N партиций для N consumer-ов

Идеальная ситуация:

Количество партиций = Количество consumer-ов

Топик: orders (4 партиции)
Consumer Group: order-processors
├── Consumer-1 обрабатывает партицию 0
├── Consumer-2 обрабатывает партицию 1
├── Consumer-3 обрабатывает партицию 2
└── Consumer-4 обрабатывает партицию 3

Результат:
- Полная параллелизация: каждый consumer работает на 100%
- Максимальная пропускная способность
- Нет конфликтов за доступ к партициям

Что происходит при неоптимальном распределении?

Случай 1: Больше consumer-ов чем партиций

Топик: orders (3 партиции)
Consumer Group: order-processors (5 consumer-ов)

├── Consumer-1 обрабатывает партицию 0 (работает)
├── Consumer-2 обрабатывает партицию 1 (работает)
├── Consumer-3 обрабатывает партицию 2 (работает)
├── Consumer-4 НЕ ПОЛУЧАЕТ НИКАКУЮ ПАРТИЦИЮ (холостой ход)
└── Consumer-5 НЕ ПОЛУЧАЕТ НИКАКУЮ ПАРТИЦИЮ (холостой ход)

Проблема: 2 consumer-а (40%) будут простаивать без работы.

Код примера:

// Consumer Group с 5 консьюмерами
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(consumerProps);
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(consumerProps);
KafkaConsumer<String, String> consumer3 = new KafkaConsumer<>(consumerProps);
KafkaConsumer<String, String> consumer4 = new KafkaConsumer<>(consumerProps); // Холостой ход
KafkaConsumer<String, String> consumer5 = new KafkaConsumer<>(consumerProps); // Холостой ход

// Все они подписаны на один топик
consumer1.subscribe(Arrays.asList("orders"));
consumer2.subscribe(Arrays.asList("orders"));
consumer3.subscribe(Arrays.asList("orders"));
consumer4.subscribe(Arrays.asList("orders"));
consumer5.subscribe(Arrays.asList("orders"));

// После балансировки:
// consumer1 получит партицию 0
// consumer2 получит партицию 1
// consumer3 получит партицию 2
// consumer4 и consumer5 не получат никаких партиций!

Случай 2: Больше партиций чем consumer-ов

Топик: orders (5 партиций)
Consumer Group: order-processors (2 consumer-а)

├── Consumer-1 обрабатывает партицию 0 И партицию 1 (двойная нагрузка)
└── Consumer-2 обрабатывает партицию 2 И партицию 3 И партицию 4 (тройная нагрузка)

Проблема: Нагрузка распределена неравномерно.

// Consumer Group с 2 консьюмерами, но топик имеет 5 партиций
KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(consumerProps);
KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(consumerProps);

consumer1.subscribe(Arrays.asList("orders"));
consumer2.subscribe(Arrays.asList("orders"));

// После балансировки:
// consumer1 получит партиции [0, 1]
// consumer2 получит партиции [2, 3, 4]

// consumer2 будет иметь 50% больше работы чем consumer1

Гарантия порядка сообщений

Ещё одна важная причина, почему партиции важны:

Каждая партиция гарантирует FIFO порядок для своих сообщений.

// Топик: user-events
// Партиция 0: сообщения для user_id=1
// Партиция 1: сообщения для user_id=2
// Партиция 2: сообщения для user_id=3

// Сообщения для одного пользователя должны идти в одну партицию
// чтобы сохранить порядок обработки

KafkaProducer<String, Event> producer = new KafkaProducer<>(producerProps);

// Используем partition key для гарантии порядка
ProducerRecord<String, Event> record1 = new ProducerRecord<>(
    "user-events",
    "user_1",  // Partition key - определяет, в какую партицию попадёт
    new Event("login")
);

ProducerRecord<String, Event> record2 = new ProducerRecord<>(
    "user-events",
    "user_1",  // Такой же ключ - попадёт в ту же партицию
    new Event("purchase")
);

ProducerRecord<String, Event> record3 = new ProducerRecord<>(
    "user-events",
    "user_1",  // Все события для user_1 в одной партиции
    new Event("logout")
);

producer.send(record1);
producer.send(record2);
producer.send(record3);

// Гарантия: события будут обработаны в порядке:
// 1. login
// 2. purchase
// 3. logout

Оптимальная архитектура

Рекомендуемый способ

// 1. Создаём топик с правильным количеством партиций
// kafka-topics.sh --create --topic orders \
//                  --partitions 10 \
//                  --replication-factor 3

// 2. Создаём consumer group с тем же количеством consumer-ов
public class OrderProcessor {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-processors");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("orders"));
        
        // Каждый запущенный экземпляр этой программы —
        // это отдельный consumer в группе "order-processors"
        
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Partition: %d, Offset: %d, Value: %s%n",
                    record.partition(), record.offset(), record.value());
                processOrder(record.value());
            }
        }
    }
}

// Запускаем 10 экземпляров этого приложения
// Kafka автоматически распределит 10 партиций между 10 consumer-ами

Масштабирование

Преимущество оптимальной архитектуры:

Начальное состояние:
- Топик: 10 партиций
- Consumer Group: 10 consumer-ов (каждый обрабатывает 1 партицию)
- Пропускная способность: 100k msg/sec

Если нужно увеличить пропускную способность в 2 раза:

Способ 1: Неправильно (не добавить партиции)
- Добавляем ещё 10 consumer-ов
- Результат: 10 consumer-ов работают, 10 холостят
- Пропускная способность: всё ещё 100k msg/sec (не увеличилась!)

Способ 2: Правильно (добавить партиции)
- Расширяем топик до 20 партиций
- Добавляем ещё 10 consumer-ов
- Результат: 20 consumer-ов работают параллельно
- Пропускная способность: 200k msg/sec (удвоилась!)

Когда можно нарушить это правило

Есть ситуации, когда один consumer обрабатывает несколько партиций:

1. Разные consumer group-ы

// Consumer Group A (для real-time обработки)
KafkaConsumer<String, String> consumerA = new KafkaConsumer<>(propsA);
consumerA.subscribe(Collections.singletonList("orders"));
// Может обрабатывать все 10 партиций одновременно

// Consumer Group B (для batch обработки)
KafkaConsumer<String, String> consumerB = new KafkaConsumer<>(propsB);
consumerB.subscribe(Collections.singletonList("orders"));
// Независимо обрабатывает все 10 партиций одновременно

2. Batch обработка (когда параллелизм не критичен)

// Для batch обработки один consumer может обработать всё
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));

// Процесс будет медленнее, но это нормально для batch
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    processBatch(records);
}

Практический пример: мониторинг

public class ConsumerLagMonitor {
    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("orders"));
        
        while (true) {
            consumer.poll(Duration.ofMillis(100));
            
            // Получаем информацию о партициях
            Set<TopicPartition> assignedPartitions = consumer.assignment();
            
            for (TopicPartition partition : assignedPartitions) {
                long committed = consumer.committed(partition).offset();
                long position = consumer.position(partition);
                long lag = position - committed;
                
                System.out.printf(
                    "Partition: %d, Lag: %d%n",
                    partition.partition(), lag
                );
            }
            
            // Здоровое состояние:
            // - Каждому consumer-у назначена 1 партиция
            // - Lag примерно одинаков для всех партиций
            // - Нет unassigned партиций
        }
    }
}

Итоговый чеклист

✓ Оптимально: количество партиций = количество consumer-ов в группе ✓ Избегай ситуаций когда consumer-ов больше чем партиций (холостой ход) ✓ Если партиций больше чем consumer-ов — нагрузка распределится неравномерно ✓ Каждая партиция гарантирует FIFO порядок сообщений ✓ Используй partition key для гарантии порядка сообщений ✓ Для масштабирования: добавляй партиции, а не consumer-ов ✓ Мониторь lag consumer-ов и распределение партиций ✓ Разные consumer group-ы могут независимо обрабатывать все партиции

Основной вывод: Каждому consumer-у должна быть назначена партиция для полного использования его вычислительного потенциала и параллельной обработки сообщений.