Share Groups (KIP-932): queue semantics в Kafka

Десятилетие Kafka жила с одним правилом: одна партиция — один consumer внутри группы. Это правило давало упорядоченность, простую модель offset и линейное масштабирование, но закрывало целый класс use-case’ов — task queues, work distribution, медленный per-message processing. Команды, которым нужны были очереди, ставили рядом RabbitMQ или SQS. KIP-932 «Queues for Kafka» закрывает этот пробел: вводит новый тип группы — share group — где несколько consumer’ов кооперативно читают из одной партиции, а acknowledgment работает на уровне отдельной записи. Preview появился в Kafka 4.0 (март 2025), GA — в Kafka 4.2 (февраль 2026).

Зачем это нужно: ограничения consumer groups

Классическая модель consumer groups устроена так: координатор группы назначает каждую партицию ровно одному consumer’у. Это даёт упорядоченное чтение в пределах партиции и гарантирует, что один offset продвигается одной сущностью. Но как только нагрузка на конкретное сообщение становится тяжёлой и неравномерной — модель ломается.

Для классического event streaming эти ограничения — фича, а не баг: они дают порядок и предсказуемость. Но для task queue, где порядок не важен, а важно «выгрести очередь как можно быстрее с retry на сбоях» — это лишние оковы.

Архитектура: broker-side delivery state

Главное архитектурное отличие share groups от consumer groups — где живёт состояние delivery. В consumer groups координация в основном клиентская: лидер группы вычисляет назначения, клиенты сами трекают offset, broker лишь хранит коммиты. В share groups broker берёт на себя гораздо больше: он отслеживает delivery state каждой отдельной записи в скользящем окне in-flight сообщений.

Скользящее окно in-flight записей живёт между двумя offset’ами:

• SPSO (Share-Partition Start Offset) — всё до этого offset уже Acknowledged или Archived;
• SPEO (Share-Partition End Offset) — верхняя граница окна; записи выше ещё не вошли в делёжку.

Окно ограничено по размеру (по умолчанию 200 записей на share-partition, регулируется group.share.record.lock.partition.limit), чтобы broker не хранил неограниченный delivery state в памяти.

Share Coordinator и __share_group_state

Состояние share group персистится в новом внутреннем топике — __share_group_state. Записи в нём ключуются по (group, topicId, partition). Share Coordinator — broker, который ведёт партицию состояния для конкретной share-partition; он же обслуживает RPC от share consumer’ов.

Coordinator пишет два типа записей:

• ShareSnapshot — полный самодостаточный снапшот состояния share-partition (SPSO, in-flight записи, их состояния и delivery counts);
• ShareUpdate — инкрементальное изменение: «такая-то запись перешла в Acknowledged», «вот это — в Archived».

При перезапуске broker’а coordinator проигрывает партицию __share_group_state от последнего snapshot и восстанавливает in-memory state. Назначения consumer’ов на partition’ы не персистятся — только эпохи, чтобы детектировать «зомби» координаторов.

Acknowledgment-модель: ACCEPT, RELEASE, REJECT, RENEW

KafkaShareConsumer работает в двух режимах: implicit (записи автоматически акают на следующем poll()) и explicit (приложение само вызывает acknowledge() для каждой записи). Explicit режим даёт три типа подтверждения:

В Kafka 4.2 (KIP-1222) добавлен четвёртый — RENEW, продление acquisition lock без изменения состояния записи. Полезно для долгих обработок, где 30-секундного дефолта не хватает, но и поднимать share.record.lock.duration.ms глобально не хочется.

Partition assignment: несколько consumer’ов на одну partition

В consumer groups действует жёсткое правило: одна partition — один consumer. В share groups оно отменено. Координатор может назначить одну и ту же partition нескольким consumer’ам того же share group — они будут разбирать её записи кооперативно, через acquisition lock.

Это меняет правила игры для масштабирования: количество worker’ов больше не привязано к количеству партиций. У тебя топик с 8 партициями и 100 worker’ов — все 100 будут активны, разбирая записи через broker-managed lock.

Версии и конфигурация

Минимальная конфигурация broker’а (Kafka 4.2):

# server.properties
group.coordinator.rebalance.protocols=classic,consumer,share
share.coordinator.state.topic.replication.factor=3
share.coordinator.state.topic.min.isr=2

Группа создаётся как share при подписке через KafkaShareConsumer — никакого ручного создания не нужно, broker сам распознаёт тип по протоколу запроса.

Ключевые tuning-параметры:

• share.record.lock.duration.ms — длительность acquisition lock (по умолчанию 30000, диапазон 1000–60000);
• group.share.record.lock.partition.limit — лимит in-flight записей на share-partition (по умолчанию 200);
• group.share.delivery.attempt.limit — максимум попыток до архивирования (по умолчанию 5);
• share.auto.offset.reset — стартовая позиция для новой share group (latest / earliest).

Code example: KafkaShareConsumer

import org.apache.kafka.clients.consumer.AcknowledgeType;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaShareConsumer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class TaskQueueWorker {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "image-processing-workers");
        props.put("key.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
        // Explicit acknowledgment: per-record control
        props.put("share.acknowledgement.mode", "explicit");

        try (KafkaShareConsumer<String, String> consumer =
                 new KafkaShareConsumer<>(props)) {
            consumer.subscribe(List.of("image-processing-tasks"));

            while (true) {
                ConsumerRecords<String, String> records =
                    consumer.poll(Duration.ofMillis(500));

                for (ConsumerRecord<String, String> record : records) {
                    try {
                        ImageJob job = parseJob(record.value());
                        processImage(job);  // долгая ML обработка
                        // Успех: подтверждаем конкретную запись
                        consumer.acknowledge(record, AcknowledgeType.ACCEPT);
                    } catch (TransientDownstreamException e) {
                        // Retry: запись вернётся в Available и достанется кому-то ещё
                        consumer.acknowledge(record, AcknowledgeType.RELEASE);
                    } catch (PoisonMessageException e) {
                        // Невосстановимо: сразу в Archived
                        consumer.acknowledge(record, AcknowledgeType.REJECT);
                    }
                }
                // Коммит acknowledgments на broker одним батчем
                consumer.commitSync();
            }
        }
    }
}

Несколько ключевых отличий от обычного KafkaConsumer:

• Нет вызова assign() или seek() — share consumer не управляет offset’ами; он только подтверждает записи.
• acknowledge(record, type) принимает конкретную запись, а не batch — это per-record acknowledgment.
• commitSync() отправляет все накопленные подтверждения на broker одной операцией, чтобы не делать round-trip на каждую запись.

Когда выбирать share groups

Типичные use case’ы под share groups:

• Task queues — обработка независимых job’ов (image resize, PDF generation, ML inference);
• Notification fan-out — рассылка уведомлений с retry на флэйки backend;
• Webhook delivery — outbound HTTP с retry и DLQ через REJECT;
• Background processing — всё, что раньше тащили в RabbitMQ или SQS «потому что Kafka так не умеет».

Limitations

Share groups — мощный инструмент, но у него есть цена.

RabbitMQ-like patterns в Kafka

С share groups Kafka впервые становится разумной заменой RabbitMQ для типичных queue use case’ов. Сравнение:

Для команд, где Kafka уже стоит как event backbone, share groups убирают главный аргумент в пользу RabbitMQ — «нам нужны очереди». Не нужно тащить второй брокер с отдельной operational нагрузкой.

Итог

Share groups (KIP-932) — крупнейшее изменение consumer-модели Kafka за десятилетие. Broker берёт на себя отслеживание delivery state каждой записи в скользящем окне; партиции назначаются нескольким consumer’ам кооперативно; per-record acknowledgment даёт ACCEPT/RELEASE/REJECT с встроенной poison message protection. Это закрывает целый класс queue use-case’ов — task queues, work distribution, retry с DLQ — для которых раньше приходилось ставить рядом RabbitMQ. Цена — отказ от ordering, дополнительная нагрузка на broker и независимость offset-пространства от обычных consumer groups. В Kafka 4.2 (февраль 2026) фича объявлена production-ready.