Могут ли несколько consumers читать одну партийцию

Могут ли несколько consumers читать одну партицию?

Ключевой момент: Да, но с ограничениями. Ответ зависит от consumer group — это концепция Kafka, которая определяет поведение.

Базовые концепции Kafka

Тема (Topic) — как очередь сообщений
  │
  ├── Партиция 0 [msg1, msg2, msg3, msg4]
  ├── Партиция 1 [msg5, msg6, msg7, msg8]
  └── Партиция 2 [msg9, msg10, msg11, msg12]

Сценарий 1: Разные Consumer Groups (МОЖНО)

// Consumer Group A
Consumer consumerA1 = new KafkaConsumer<>(propsA);
consumerA1.subscribe(Collections.singletonList("my-topic"));

// Consumer Group B
Consumer consumerB1 = new KafkaConsumer<>(propsB);
consumerB1.subscribe(Collections.singletonList("my-topic"));

// propsA.put("group.id", "group-a");
// propsB.put("group.id", "group-b");

Результат:

Тема: my-topic
  ├── Партиция 0: Group A (может читать) + Group B (может читать)
  ├── Партиция 1: Group A (может читать) + Group B (может читать)
  └── Партиция 2: Group A (может читать) + Group B (может читать)

Каждая группа получает ВСЕ сообщения!

✅ Можно — каждая группа читает независимо

Сценарий 2: Один Consumer Group (НЕЛЬЗЯ)

// Consumer 1
props1.put("group.id", "my-group");
Consumer consumer1 = new KafkaConsumer<>(props1);
consumer1.subscribe(Collections.singletonList("my-topic"));

// Consumer 2
props2.put("group.id", "my-group");  // ТА ЖЕ группа!
Consumer consumer2 = new KafkaConsumer<>(props2);
consumer2.subscribe(Collections.singletonList("my-topic"));

Результат:

Тема: my-topic
  ├── Партиция 0: Consumer 1 ИЛИ Consumer 2
  ├── Партиция 1: Consumer 1 ИЛИ Consumer 2
  └── Партиция 2: Consumer 1 ИЛИ Consumer 2

Каждую партицию читает только ОДИН consumer!

❌ Нельзя — Kafka распределяет партиции между consumers в группе

Механизм: Consumer Group Rebalancing

// Consumer Group: payment-processors
// Тема: orders (3 партиции)

// Шаг 1: Запускаем Consumer 1
Consumer c1 = new KafkaConsumer<>(props);
c1.subscribe(Collections.singletonList("orders"));

// Rebalance:
// c1 → получает все партиции [0, 1, 2]

// Шаг 2: Запускаем Consumer 2
Consumer c2 = new KafkaConsumer<>(props);
c2.subscribe(Collections.singletonList("orders"));

// Rebalance:
// c1 → [0, 1]     (перераспределение!)
// c2 → [2]

// Шаг 3: Запускаем Consumer 3
Consumer c3 = new KafkaConsumer<>(props);
c3.subscribe(Collections.singletonList("orders"));

// Rebalance:
// c1 → [0]
// c2 → [1]
// c3 → [2]

// Шаг 4: Если выходит Consumer 2
// Rebalance:
// c1 → [0, 1]
// c3 → [2]

Код: Реальный пример

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-processing-group");
        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(Arrays.asList("orders"));
        
        // Этот consumer прочитает свою часть партиций
        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());
            }
        }
    }
}

// Если запустить 3 экземпляра этого класса с group.id="order-processing-group"
// Каждый получит разные партиции

Как работает распределение

Стратегия: RangeAssignor (по умолчанию)

Тема: orders (6 партиций)
Consumers в группе: 2

Распределение:
  Consumer 1: [0, 1, 2]  (первые 3)
  Consumer 2: [3, 4, 5]  (остальные 3)

Стратегия: RoundRobinAssignor

Распределение:
  Consumer 1: [0, 2, 4]
  Consumer 2: [1, 3, 5]

Когда это полезно?

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

// Одна партиция = 100 msg/sec
// Нужно обработать 300 msg/sec
// Решение: 3 consumers в одной группе

// Каждый обрабатывает 100 msg/sec
// Вместе = 300 msg/sec ✅

// КОД:
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("orders"));
        // Каждый thread получит одну партицию
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            records.forEach(record -> processOrder(record));
        }
    }).start();
}

Важно: Порядок гарантий

// ❌ ПОРЯДОК ГАРАНТИРОВАН только в одной партиции
// Сообщения 1, 2, 3 в партиции 0 → всегда в этом порядке
// Сообщения 4, 5, 6 в партиции 1 → всегда в этом порядке
// НО: 1, 2, 3, 4, 5, 6 — порядок между партициями НЕ гарантирован

// ✅ Если нужен порядок всех сообщений:
// Используйте ОДНУ партицию (тогда нельзя масштабировать)
// ИЛИ используйте одинаковый ключ для связанных сообщений

public void sendOrders() throws Exception {
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    
    // Все заказы от user1 → одна партиция (через key)
    producer.send(new ProducerRecord<>("orders", "user1", "order-1"));
    producer.send(new ProducerRecord<>("orders", "user1", "order-2"));
    producer.send(new ProducerRecord<>("orders", "user1", "order-3"));
    // Все идут в одну партицию → порядок сохранен
}

Частые ошибки

Ошибка 1: Две партиции — одному consumer

// ❌ НЕПРАВИЛЬНО
Consumer consumer = new KafkaConsumer<>(props);
consumer.assign(Arrays.asList(
    new TopicPartition("orders", 0),
    new TopicPartition("orders", 1)  // Одному consumer две партиции
));

// ✅ ПРАВИЛЬНО (если это нужно)
consumer.assign(Arrays.asList(
    new TopicPartition("orders", 0)
));
// Вторую партицию обработает другой consumer

Ошибка 2: Забыли про rebalancing

// Rebalancing может занять время
// В это время consumer не обрабатывает данные
// Не забывайте про session.timeout.ms

props.put("session.timeout.ms", "30000");  // 30 сек
props.put("heartbeat.interval.ms", "10000");  // 10 сек heartbeat

Итог

Могут ли несколько consumers читать одну партицию в Kafka?

✅ ДА, если они в разных consumer groups

• Каждая группа независимо читает все сообщения
• Могут быть в разных состояниях (offset)

❌ НЕТ, если они в одной consumer group

• Kafka распределяет партиции между consumers
• Каждую партицию читает только один consumer
• Максимум consumers = количество партиций
• При добавлении/удалении consumers → rebalancing

Правило: 1 Partition = 1 Consumer (в одной группе)