В прошлый раз событие читал один consumer. Но магазин растёт: каждый заказ порождает события OrderCreated, OrderPaid и OrderShipped, и сервис уведомлений уже не успевает рассылать письма. Хочется запустить несколько копий сервиса и поделить работу так, чтобы ни одно событие не обработали дважды. К тому же те же события нужны аналитике, и она не должна мешать письмам. Всё это решают группы потребителей. А ещё выясним, что происходит при сбое между обработкой и сохранением закладки, и как перечитать события заново.

Группа потребителей

Все consumer с одинаковым groupId образуют группу. Kafka делит партиции топика между участниками группы так, что у каждой партиции в каждый момент ровно один читатель из группы. Аналогия: три кассы в магазине (партиции) и кассиры (consumers). Каждую кассу обслуживает один кассир, иначе очередь покупателей перемешается; один кассир может вести две кассы, а лишний кассир остаётся без кассы.

Consumers в группеКто что читает (пример для 3 партиций)
1A: партиции 0, 1, 2
2A: 0 и 2, B: 1
3A: 0, B: 1, C: 2
4трое читают по одной, четвёртый простаивает
Три партиции делятся между consumers одной группы; когда подключается третий, происходит ребаланс и каждый получает по партиции
Три партиции делятся между consumers одной группы; когда подключается третий, происходит ребаланс и каждый получает по партиции

Отсюда правило масштабирования: читать параллельно можно не больше, чем партиций. Хотите восемь копий сервиса — нужно минимум восемь партиций. Вот consumer, который показывает, какие партиции ему достались:

js
import { Kafka } from 'kafkajs';

const name = process.argv[2] ?? 'A';
const kafka = new Kafka({
  clientId: `notifications-${name}`,
  brokers: ['localhost:9092'],
});
const consumer = kafka.consumer({ groupId: 'notifications' });

consumer.on(consumer.events.GROUP_JOIN, ({ payload }) => {
  const mine = payload.memberAssignment.orders ?? [];
  console.log(name, 'got partitions', mine);
});

await consumer.connect();
await consumer.subscribe({ topic: 'orders', fromBeginning: true });
await consumer.run({
  eachMessage: async ({ partition, message }) => {
    const event = JSON.parse(message.value.toString());
    console.log(name, `p${partition}`, event.type, event.orderId);
  },
});

Запустите его в двух терминалах: node group.js A и node group.js B. Вывод может выглядеть так (какой consumer получит какие партиции, зависит от алгоритма; служебные JSON-строки kafkajs опущены):

text
A got partitions [ 0, 2 ]
B got partitions [ 1 ]
A p2 OrderCreated 3
B p1 OrderCreated 4
A p0 OrderCreated 5

Запустите третий — группа перераспределит партиции:

text
A got partitions [ 0 ]
B got partitions [ 1 ]
C got partitions [ 2 ]

Ребаланс