Что такое consumer group в Kafka и как она помогает разделять нагрузку между потребителями?
Что проверяют
Понимаете ли вы, что единица параллелизма в Kafka это партиция, а не консьюмер, и как это ложится на горутины. Следом почти всегда спрашивают, что будет, если консьюмеров больше, чем партиций, и когда коммитить оффсет.
Как отвечать
Consumer group это набор консьюмеров с одинаковым group.id, которые делят между собой партиции топика:
- распределение: каждую партицию в группе читает ровно один консьюмер, один консьюмер может держать несколько партиций. Порядок гарантирован только внутри партиции;
- предел масштабирования: 12 партиций дают максимум 12 активных консьюмеров, тринадцатый будет простаивать;
- разные группы: каждая группа получает все сообщения топика и ведёт свои оффсеты;
- оффсеты: хранятся в служебном топике
__consumer_offsets, после рестарта группа продолжает с последнего закоммиченного; - ребалансировка: при входе, выходе или падении участника координатор перераздаёт партиции. В классическом протоколе eager-стратегии останавливают всю группу,
cooperative-stickyдвигает только нужные партиции, а новый протокол KIP-848 (group.protocol=consumer, стабилен с Kafka 4.0) переносит назначение на брокер.
В Go обычно берут один из клиентов: segmentio/kafka-go (kafka.NewReader с GroupID), IBM/sarama (ConsumerGroupHandler, где ConsumeClaim крутится в своей горутине на каждую партицию), twmb/franz-go или confluent-kafka-go поверх librdkafka. Для at-least-once коммитят после обработки: в kafka-go это пара FetchMessage и CommitMessages, а ReadMessage при заданном GroupID коммитит сам.
Где спотыкаются
Запускают горутину на каждое сообщение. Порядок теряется, а коммит оффсета N означает «всё до N обработано», и при падении часть сообщений пропадёт. Параллелят по партициям или по ключу.
Добавляют поды консьюмера, не увеличив число партиций, и удивляются, что пропускная способность не выросла.
Держат долгую обработку, не реагируя на ребалансировку. В Sarama ConsumeClaim обязан выйти, когда закрыт claim.Messages() или отменён session.Context(), в librdkafka то же сторожит max.poll.interval.ms. Иначе участника выкидывают из группы, и сообщения обрабатываются повторно.