Consumer group, rebalance и commit offset

Группа — способ разделить чтение одного топика. Координатор назначает каждой партиции ровно одного члена группы. Два процесса с одним group.id делят работу.

Два процесса с разными group.id — две независимые подписки. Каждая читает лог со своим курсором. Это fan-out: оба получают все сообщения. Это не competing consumers как в Rabbit, где воркеры делят задачи из одной очереди.

Пример. Биллинг и поиск читают топик orders. Разные group.id — каждый получит все заказы. Один group.id у двух инстансов биллинга — они поделят партиции, каждый заказ обработает только один инстанс.

Offset — курсор, не удаление

Offset — позиция «я обработал досюда» в партиции. В современных кластерах он лежит в служебном топике __consumer_offsets, не в ZooKeeper.

Commit offset — запись курсора. Сообщение из лога при этом не удаляется. Оно останется до retention. Другая группа прочитает его со своего места: с начала, с конца или с выбранного сдвига.

Когда коммитить

От момента commit зависит, что будет при падении. Сначала работа, потом commit — при крэше обработаете ещё раз. Это at-least-once: сообщение не потеряется, но может прийти повторно.

Сначала commit, потом работа — при крэше потеряете. Это at-most-once. Автокоммит «по таймеру, не глядя» даёт то или другое в зависимости от момента падения. Получаются и дыры, и дубли.

for {
	msg, err := consumer.Fetch(ctx)
	if err != nil {
		return err
	}
	if err := process(ctx, msg); err != nil {
		return err // offset не коммитим: сообщение придёт снова
	}
	if err := consumer.Commit(ctx, msg); err != nil {
		return err
	}
}
// commit до process = at-most-once: крэш после commit — сообщение потеряно

Пример. Оплатили заказ, упали до commit. После рестарта то же сообщение придёт снова. Без проверки «этот заказ уже оплачен» снимут деньги дважды.

Rebalance

Член вошёл, вышел, не успел poll за max.poll.interval.ms, изменилась подписка — координатор переназначает партиции. Это rebalance.

На время eager rebalance все члены могут остановить обработку. Cooperative sticky забирает только те партиции, которые надо отдать. Это мягче, но сюрприз на границе остаётся.

Пока ребаланс идёт, легко обработать сообщение дважды. Старый владелец мог обработать, но не закоммитить. Новый начинает с последнего закоммиченного offset и повторяет работу. Если побочный эффект нельзя делать дважды, нужен идемпотентный consumer.

Долгий process опаснее, чем кажется. Если он длится дольше max.poll.interval, член считается мёртвым. Партиции уезжают. Вы всё ещё работаете над сообщением — а сосед уже взял тот же offset. Отсюда ребаланс по кругу и дубли.

Пример. Обработка одного сообщения занимает 8 минут, max.poll.interval — 5 минут. Координатор решает, что вы умерли, отдаёт партицию соседу. Оба обрабатывают одно и то же.

Кому какой group.id

Несколько инстансов сервиса с одним group.id — горизонтальное масштабирование одной обработки. Те же инстансы с разными id — умножение side effect: каждый заказ обработается столько раз, сколько инстансов.

Частая причина разных id — случайный идентификатор в деве, который утекает в прод, или опечатка в конфиге. Compacted топик это не лечит: compaction ужимает лог, а не обещает ровно один вызов вашей БД.