Skip to main content
Java

Queues for Kafka: Share Groups with Spring for Apache Kafka 4

Kafka share groups (KIP-932) let many consumers share one partition and acknowledge records one by one. Measured against a real Kafka 4.3.1 broker with Spring Kafka 4.1.1: what you gain, what you lose, and when RabbitMQ is still the answer.

Ankur 9 min read
Kafka’s consumer groups have one rule that frustrates people who just want a work queue: inside a group, a partition belongs to exactly one consumer. Ten consumers on a two-partition topic means eight sit idle. Share groups, introduced by KIP-932 and nicknamed “Queues for Kafka”, remove that rule: many consumers can read the same partition, and each record is acknowledged individually. Spring for Apache Kafka already supports them. This post runs share groups against a real broker, and the first thing it shows is that “removes the rule” needs a footnote.
Versions and limits. Apache Kafka broker 4.3.1 (single node, KRaft, on localhost), kafka-clients 4.2.1 (managed by Boot), Spring Kafka 4.1.1 (GA) on Spring Boot 4.1.1, JDK 25 (Temurin). On this broker the share.version feature was already finalized at level 1 and I set no feature flag. One broker, small topics: this shows behaviour, not throughput. Every output below is a real run; timings vary between runs, so trust the shape and not the milliseconds. All code and raw output is in the share-groups module of my spring-messaging-demo repository.

The failure first: six consumers, two partitions, and still only two worked

The test starts six consumers on a two-partition topic and gives them 60 records that each take 50 ms to process. I ran it three ways (01-work-spread.txt): 
consumer group                         : consumers that received records = 2
share group, default share.acquire.mode: consumers that received records = 2
share group, acquire.mode=record_limit : consumers that received records = 6

consumer group:      26 and 34 records on two consumers, 1993 ms
share, default:      26 and 34 records on two consumers, 1778 ms
share, record_limit: 8 to 12 records on each of six, 681 ms
With the default setting a share group behaved just like a consumer group: two consumers did all the work. Only when I set the consumer config share.acquire.mode=record_limit did all six share the load, and the run finished in roughly a third of the time. The default mode, batch_optimized, hands out whole batches, so with few records whole batches go to the first consumers that ask. The default mode also ignored max.poll.records in my runs; record_limit is the mode where I used it (set to 3). If you adopt share groups to scale past the partition count and keep the default, you will see no benefit and wonder why.
Consumer group2 of 6 Share group (default)2 of 6 Share group (record_limit)6 of 6
The chart is the three summary lines above, with bars scaled to six consumers.

What a share group is

In a consumer group the broker assigns partitions to consumers. In a share group the broker instead lends individual records: a consumer fetches records, they are locked to it for a while, and it tells the broker what happened to each one. Three answers exist:
  • ACCEPT: processed, done.
  • RELEASE: give it back; another delivery will follow.
  • REJECT: this record is bad, do not deliver it again.
The client API has a fourth, RENEW, which extends the lock; I did not test it. Because acknowledgement is per record there is no committed offset to rewind to, and that choice explains most of the surprises below.

Acknowledgement types, as the broker delivers them

Four records through the raw share consumer (02-ack-types.txt): 
ok              delivery #1 -> accept
retry-once      delivery #1 -> release
reject-me       delivery #1 -> reject
always-release  delivery #1 -> release
retry-once      delivery #2 -> accept
always-release  delivery #2 -> release
always-release  delivery #3 -> release
always-release  delivery #4 -> release
always-release  delivery #5 -> release

deliveries of ok             = 1
deliveries of retry-once     = 2
deliveries of reject-me      = 1
deliveries of always-release = 5
The delivery number comes from ConsumerRecord.deliveryCount(). A record released every time was delivered five times and then not again, so on this broker the delivery attempt limit behaves as 5. I read that off the output; I did not look up or change the broker setting behind it, and I did not look for what happens to the record afterwards, so there is no dead-letter story here. If you need one, plan to record the delivery count yourself and divert the record before the limit.

A new share group sees nothing

I produced three records before any group existed (03-share-group-starts-at-latest.txt): 
produced 3 records before any group existed
new group, default share.auto.offset.reset      -> received 0 records
new group, share.auto.offset.reset=earliest      -> received 3 records
A new share group starts at the latest offset, so records produced before its first consumer joined are skipped. The setting share.auto.offset.reset is a group config, not a consumer property: I set it with the Admin client’s incrementalAlterConfigs on a resource of type GROUP. A consumer property of that name would not do it. Plan this step into your deployment, or the first messages after a new queue is created can vanish.

You give up ordering

Forty records with the same key on one partition, three consumers of different speed, produced in order 0 to 39 (04-no-ordering-guarantee.txt): 
processed  : [4, 5, 6, 7, 12, 0, 13, 14, 15, 16, 17, 18, 19, 1, 20, 8, 21, ... 35, 11]
adjacent pairs that arrived out of order = 9
This is the price of sharing a partition. Record 0 finished after record 12. Even a key does not help: a share group does not keep same-key records on one consumer. If your records for one customer must be applied in order, share groups are the wrong tool; keep a consumer group and partition by key. The count of 9 out-of-order pairs is from one run and will differ on yours.

Spring for Apache Kafka 4.1

Spring Kafka 4.1.1 already contains the share classes: ShareKafkaListenerContainerFactory, DefaultShareConsumerFactory, ShareAcknowledgment (with acknowledge, release, reject and renew), ContainerProperties.ShareAckMode with IMPLICIT, EXPLICIT and MANUAL, and ShareConsumerRecordRecoverer with REJECTING and RELEASING variants. Spring Boot 4.1.1 has no auto-configuration for them, so you declare the beans yourself. This is ShareConfig.java, abridged: 
@Bean
ShareConsumerFactory<String, String> shareConsumerFactory(@Value("${broker:localhost:9092}") String bootstrap) {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put("share.acquire.mode", "record_limit");
    props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "3");
    return new DefaultShareConsumerFactory<>(props);
}

@Bean
ShareKafkaListenerContainerFactory<String, String> explicitShareFactory(ShareConsumerFactory<String, String> factory) {
    ShareKafkaListenerContainerFactory<String, String> f = new ShareKafkaListenerContainerFactory<>(factory);
    f.setConcurrency(2);
    f.getContainerProperties().setShareAckMode(ShareAckMode.MANUAL);
    return f;
}
With that in place the listener takes a ShareAcknowledgment and decides per record: 
@KafkaListener(id = "manual", topics = "${share.topic.manual}", groupId = "${share.group.manual}",
        containerFactory = "explicitShareFactory")
void manual(ConsumerRecord<String, String> record, ShareAcknowledgment ack) {
    int attempt = record.deliveryCount().orElse((short) 1);
    switch (record.value()) {
        case "bad"   -> ack.reject();
        case "again" -> { if (attempt == 1) ack.release(); else ack.acknowledge(); }
        default      -> ack.acknowledge();
    }
}
The real output through Spring’s container (05-spring-share-listeners.txt): 
== manual mode (ShareAckMode.MANUAL), listener takes a ShareAcknowledgment
again delivery #1 -> release
again delivery #2 -> acknowledge
bad delivery #1 -> reject
ok delivery #1 -> acknowledge
One more result worth knowing. In implicit mode I set concurrency to 4 and sent 24 records to a one-partition topic. Only two listener threads handled records: 
== implicit mode, concurrency 4, 24 records on one topic partition
records handled = 24
distinct listener threads that handled records = 2
Concurrency in the container is an upper bound; the broker decides how many consumers actually receive records. Do not size your thread pool from the concurrency setting alone, and measure.
Consumer group partition 0 consumer A Share group partition 0 consumer A consumer B each record acked on its own
The drawing is the idea in one picture: one partition feeding one consumer on top, and one partition feeding several below, which is why order is lost.
Run it yourself 
git clone https://ankurm.com/git.app/asmhatre/spring-messaging-demo.git
cd spring-messaging-demo/share-groups
scripts/broker-up.sh      # downloads Kafka 4.3.1, checks its checksum, starts KRaft on localhost:9092
scripts/run-all.sh        # runs the five tests and rewrites output/*.txt
scripts/broker-down.sh
The tests do not start a broker. Each creates its own uniquely named topic and group, so reruns do not see old records.
Two consumer configs that did the work share.acknowledgement.mode (implicit or explicit) controls whether the client acknowledges for you; Spring’s ShareAckMode sits on top of it. share.acquire.mode is batch_optimized (the default I observed) or record_limit. I used only the second to change behaviour.

Share groups or RabbitMQ?

This section is judgement, built on the results above plus the broader comparison in Kafka vs RabbitMQ vs Pulsar; I did not benchmark RabbitMQ here.
  • Choose share groups when you already run Kafka, the work is independent per record, a slow record should not hold up the ones behind it, and you want more workers than partitions.
  • Keep consumer groups when order per key matters.
  • Still choose RabbitMQ when you need a broker built around queues from the start: routing by exchange and binding, per-message features, and a long-established dead-letter pattern. In this run I found a delivery limit but no dead-letter behaviour, and the Spring side needed hand-written beans. If you want a message to leave and be gone, RabbitMQ gets you there with less to assemble.
If you do use Kafka consumer groups with retries and dead-letter topics, see Spring Kafka 4.1 error handling. Coming from Spring Kafka 3? Read the 3 to 4 migration guide.
Should you even do this? Only if the queue semantics are the reason. If your consumers are the bottleneck because they wait on slow I/O, a consumer group with more partitions, or virtual threads inside one consumer, is simpler and keeps your ordering. Share groups cost you ordering, give you a fixed delivery limit, need a group-level offset setting, and in Spring Boot 4.1 need manual configuration. They are worth it when you have been bending consumer groups into a work queue and fighting the partition count.

Further reading

No Comments yet!

Leave a Reply

This site uses Akismet to reduce spam. Learn how your comment data is processed.