diff --git a/README.md b/README.md
index 4d24147..9b9f8bc 100644
--- a/README.md
+++ b/README.md
@@ -3,8 +3,8 @@
Companion code for the messaging series on [ankurm.com](https://ankurm.com). Each directory is a
self-contained Maven project for one article, with its own `pom.xml` and its own captured
output, regenerated from the test suite rather than typed by hand. Most modules keep that
-output under `docs/output/` alongside numbered documentation chapters; `saga/` keeps it directly
-under `output/` instead, with the equivalent depth in the article's own accordion sections.
+output under `docs/output/` alongside numbered documentation chapters; `saga/` and `share-groups/` keep it
+directly under `output/` instead, with the equivalent depth in the article's own accordion sections.
| Module | Article | What it demonstrates |
|---|---|---|
@@ -15,6 +15,7 @@ under `output/` instead, with the equivalent depth in the article's own accordio
| [`protocol-comparison/`](protocol-comparison/README.md) | [RSocket vs gRPC vs WebSocket on Spring Boot 4.1: When Each One Wins](https://ankurm.com/rsocket-vs-grpc-vs-websocket-spring-boot-4-1/) | One application serving the same two operations over all three, benchmarked in one JVM, ending in the only measurement that transfers off the box: what each server produces for a consumer that asked for a hundred |
| [`broker-comparison/`](broker-comparison/README.md) | [Kafka vs RabbitMQ vs Pulsar for Java Teams: A Decision Framework with Benchmarks](https://ankurm.com/kafka-vs-rabbitmq-vs-pulsar-java-decision-framework/) | The three brokers measured side by side on ordering, replay, consumer scaling and operational footprint, ending in a decision table where every row has a transcript behind it |
| [`saga/`](saga/README.md) | Saga Pattern in Spring Boot: Orchestration vs Choreography with Kafka | The same order/payment/inventory saga built twice against a real broker — choreography and orchestration — driven through the same inventory failure so the compensating transaction can be compared side by side, plus a real duplicate-delivery test for the idempotent-consumer problem |
+| [`share-groups/`](share-groups/README.md) | [Queues for Kafka: Share Groups with Spring for Apache Kafka 4](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/) | Share groups against a real Kafka 4.3.1 broker: six consumers on two partitions, the acquire mode that decides whether they really share, the three acknowledgement types, a group that starts at the end of the topic, and the ordering you give up |
The brokers make an instructive set. Kafka's consumer holds an offset and the broker remembers
nothing about individual records; RabbitMQ's broker owns the message until it is acknowledged and
diff --git a/share-groups/.gitignore b/share-groups/.gitignore
new file mode 100644
index 0000000..a7b0fe2
--- /dev/null
+++ b/share-groups/.gitignore
@@ -0,0 +1,2 @@
+target/
+.kafka/
diff --git a/share-groups/README.md b/share-groups/README.md
new file mode 100644
index 0000000..a8eeb5a
--- /dev/null
+++ b/share-groups/README.md
@@ -0,0 +1,59 @@
+# share-groups
+
+Companion code for *Queues for Kafka: Share Groups with Spring for Apache Kafka 4* on
+[ankurm.com](https://ankurm.com/kafka-share-groups-spring-kafka-4-queues-for-kafka/).
+
+What a Kafka **share group** (KIP-932) does differently from a consumer group, measured against a
+real broker: how many consumers can share one partition, what the three acknowledgement types do to a
+record, why a new share group sees nothing, and what ordering you give up. The same ideas are then run
+through Spring Kafka's share containers.
+
+## Versions
+
+| Artifact | Version |
+|---|---|
+| Spring Boot | 4.1.1 (GA; `4.2.0-M2` is a milestone) |
+| Spring Kafka | 4.1.1, managed by the Boot BOM (share consumer support is already in this GA release) |
+| kafka-clients | 4.2.1, managed by the Boot BOM |
+| Kafka broker | 4.3.1, started by `scripts/broker-up.sh` (`share.version` is finalized at level 1 by default; no feature flag was set) |
+| JDK | 25 (LTS) |
+
+## Quickstart
+
+```
+scripts/broker-up.sh # downloads Kafka 4.3.1 (checksum-verified), starts KRaft on localhost:9092
+scripts/run-all.sh # runs every test and rewrites output/*.txt
+scripts/broker-down.sh
+```
+
+The tests do not start a broker themselves. Use `-Dbroker=host:port` if yours is elsewhere. Each test
+creates its own uniquely named topic and group, so reruns never see earlier records.
+
+## What each test is
+
+| Test | Role |
+|---|---|
+| `WorkSpreadTest` | 6 consumers, a 2-partition topic, 60 records at 50 ms each: consumer group, share group with the default `share.acquire.mode`, and share group with `record_limit` |
+| `AckTypesTest` | ACCEPT, RELEASE and REJECT through the raw `KafkaShareConsumer`, including a record that is released every time |
+| `OffsetResetTest` | A new share group starts at the latest offset; `share.auto.offset.reset=earliest` is a group config set through the Admin API |
+| `OrderingTest` | One key, one partition, three consumers of different speed: arrival order is not produce order |
+| `SpringShareListenerTest` | `ShareKafkaListenerContainerFactory` in implicit and manual acknowledgement modes with `@KafkaListener` |
+
+`ShareConfig` is the hand-written configuration: Spring Boot 4.1.1 has no auto-configuration for share
+consumers.
+
+## Captured output (`output/`)
+
+| File | What it captures |
+|---|---|
+| `01-work-spread.txt` | Which of six consumers received records under each of the three setups, with indicative timings |
+| `02-ack-types.txt` | Every delivery of four records: accepted, released once, rejected, released until Kafka stops redelivering |
+| `03-share-group-starts-at-latest.txt` | 3 records produced first: 0 received by default, 3 received after switching the group to earliest |
+| `04-no-ordering-guarantee.txt` | The order 40 same-key records were processed in, and the number of out-of-order neighbours |
+| `05-spring-share-listeners.txt` | Listener threads under implicit mode, and per-record release, reject and acknowledge under manual mode |
+
+Timings and the exact split of records between consumers differ from run to run. The shape of each
+result is stable; the milliseconds are not.
+
+No `docs/` chapter directory in this repository -- the intermediate and reference-depth material is in
+accordion sections inside the WordPress post.
diff --git a/share-groups/output/01-work-spread.txt b/share-groups/output/01-work-spread.txt
new file mode 100644
index 0000000..b6730e9
--- /dev/null
+++ b/share-groups/output/01-work-spread.txt
@@ -0,0 +1,29 @@
+topic partitions = 2, consumers = 6, records = 60, work per record = 50 ms
+== consumer group (6 KafkaConsumers, same group.id)
+consumer-0 processed 26 records
+consumer-1 processed 34 records
+consumer-2 processed 0 records
+consumer-3 processed 0 records
+consumer-4 processed 0 records
+consumer-5 processed 0 records
+total processed = 60 in 1993 ms
+== share group (6 ShareConsumers, default share.acquire.mode)
+consumer-0 processed 26 records
+consumer-1 processed 34 records
+consumer-2 processed 0 records
+consumer-3 processed 0 records
+consumer-4 processed 0 records
+consumer-5 processed 0 records
+total processed = 60 in 1778 ms
+== share group (6 ShareConsumers, share.acquire.mode=record_limit)
+consumer-0 processed 9 records
+consumer-1 processed 12 records
+consumer-2 processed 10 records
+consumer-3 processed 8 records
+consumer-4 processed 9 records
+consumer-5 processed 12 records
+total processed = 60 in 681 ms
+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
+(elapsed times vary from run to run; the shape, not the milliseconds, is the result)
diff --git a/share-groups/output/02-ack-types.txt b/share-groups/output/02-ack-types.txt
new file mode 100644
index 0000000..a965347
--- /dev/null
+++ b/share-groups/output/02-ack-types.txt
@@ -0,0 +1,13 @@
+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
diff --git a/share-groups/output/03-share-group-starts-at-latest.txt b/share-groups/output/03-share-group-starts-at-latest.txt
new file mode 100644
index 0000000..4f527bb
--- /dev/null
+++ b/share-groups/output/03-share-group-starts-at-latest.txt
@@ -0,0 +1,3 @@
+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
diff --git a/share-groups/output/04-no-ordering-guarantee.txt b/share-groups/output/04-no-ordering-guarantee.txt
new file mode 100644
index 0000000..b331fea
--- /dev/null
+++ b/share-groups/output/04-no-ordering-guarantee.txt
@@ -0,0 +1,3 @@
+produced : 0..39 in order, one key, one partition
+processed : [4, 5, 6, 7, 12, 0, 13, 14, 15, 16, 17, 18, 19, 1, 20, 8, 21, 22, 23, 24, 2, 25, 26, 27, 28, 29, 30, 3, 31, 36, 9, 37, 38, 39, 32, 33, 10, 34, 35, 11]
+adjacent pairs that arrived out of order = 9
diff --git a/share-groups/output/05-spring-share-listeners.txt b/share-groups/output/05-spring-share-listeners.txt
new file mode 100644
index 0000000..b370afb
--- /dev/null
+++ b/share-groups/output/05-spring-share-listeners.txt
@@ -0,0 +1,10 @@
+== implicit mode, concurrency 4, 24 records on one topic partition
+records handled = 24
+distinct listener threads that handled records = 2
+ implicit-C-1
+ implicit-C-2
+== manual mode (ShareAckMode.MANUAL), listener takes a ShareAcknowledgment
+again delivery #1 -> release
+again delivery #2 -> acknowledge
+bad delivery #1 -> reject
+ok delivery #1 -> acknowledge
diff --git a/share-groups/pom.xml b/share-groups/pom.xml
new file mode 100644
index 0000000..32d0134
--- /dev/null
+++ b/share-groups/pom.xml
@@ -0,0 +1,52 @@
+
+ 4.0.0
+
+
+
+ org.springframework.boot
+ spring-boot-starter-parent
+ 4.1.1
+
+
+
+ com.ankurm
+ share-groups
+ 1.0
+ jar
+
+
+ 25
+ UTF-8
+
+
+
+
+ org.springframework.boot
+ spring-boot-starter
+
+
+ org.springframework.boot
+ spring-boot-starter-kafka
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+ org.awaitility
+ awaitility
+ test
+
+
+
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+
+
+
diff --git a/share-groups/scripts/broker-down.sh b/share-groups/scripts/broker-down.sh
new file mode 100755
index 0000000..d469ba3
--- /dev/null
+++ b/share-groups/scripts/broker-down.sh
@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+# Stops the broker started by broker-up.sh (matched by its main class, not by a broad pattern).
+for p in $(ps -eo pid,cmd | grep '[k]afka.Kafka' | awk '{print $1}'); do kill "$p"; done
diff --git a/share-groups/scripts/broker-up.sh b/share-groups/scripts/broker-up.sh
new file mode 100755
index 0000000..a3b6dd0
--- /dev/null
+++ b/share-groups/scripts/broker-up.sh
@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+# Downloads Apache Kafka 4.3.1 (checksum-verified), formats a single-node KRaft cluster and starts it.
+# Usage: scripts/broker-up.sh [client-port] (default 9092; the controller uses client-port + 1)
+set -euo pipefail
+
+VERSION=4.3.1
+PORT="${1:-9092}"
+CTRL=$((PORT + 1))
+HOME_DIR="${KAFKA_DIR:-$(cd "$(dirname "$0")/.." && pwd)/.kafka}"
+TGZ="kafka_2.13-${VERSION}.tgz"
+mkdir -p "$HOME_DIR" && cd "$HOME_DIR"
+
+if [ ! -d "kafka_2.13-${VERSION}" ]; then
+ curl -fsSL -o "$TGZ" "https://downloads.apache.org/kafka/${VERSION}/${TGZ}"
+ curl -fsSL -o "${TGZ}.sha512" "https://downloads.apache.org/kafka/${VERSION}/${TGZ}.sha512"
+ # The .sha512 file is "name: HEX HEX HEX ..." in groups, so compare after stripping spaces and the name.
+ expected="$(sed 's/^[^:]*: *//' "${TGZ}.sha512" | tr -d ' \n' | tr 'A-F' 'a-f')"
+ actual="$(sha512sum "$TGZ" | cut -d' ' -f1)"
+ [ "$expected" = "$actual" ] || { echo "checksum mismatch for $TGZ" >&2; exit 1; }
+ tar -xzf "$TGZ"
+fi
+
+cd "kafka_2.13-${VERSION}"
+sed -i "s#^listeners=.*#listeners=PLAINTEXT://:${PORT},CONTROLLER://:${CTRL}#" config/server.properties
+sed -i "s#^advertised.listeners=.*#advertised.listeners=PLAINTEXT://localhost:${PORT}#" config/server.properties
+sed -i "s#^controller.quorum.bootstrap.servers=.*#controller.quorum.bootstrap.servers=localhost:${CTRL}#" config/server.properties
+sed -i "s#^log.dirs=.*#log.dirs=${HOME_DIR}/data-${PORT}#" config/server.properties
+
+if [ ! -d "${HOME_DIR}/data-${PORT}" ]; then
+ bin/kafka-storage.sh format --standalone -t "$(bin/kafka-storage.sh random-uuid | tail -1)" -c config/server.properties
+fi
+export KAFKA_HEAP_OPTS="${KAFKA_HEAP_OPTS:--Xmx512m -Xms512m}"
+setsid nohup bin/kafka-server-start.sh config/server.properties > "${HOME_DIR}/broker-${PORT}.log" 2>&1 < /dev/null &
+
+for _ in $(seq 1 60); do
+ bin/kafka-topics.sh --bootstrap-server "localhost:${PORT}" --list > /dev/null 2>&1 && break
+ sleep 2
+done
+echo "broker up on localhost:${PORT}; share.version feature:"
+bin/kafka-features.sh --bootstrap-server "localhost:${PORT}" describe | grep share.version
diff --git a/share-groups/scripts/run-all.sh b/share-groups/scripts/run-all.sh
new file mode 100755
index 0000000..9dbfd1e
--- /dev/null
+++ b/share-groups/scripts/run-all.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+# Regenerates every file in output/ from the tests. Needs a broker: scripts/broker-up.sh [port], then
+# pass -Dbroker=localhost: if it is not 9092.
+set -euo pipefail
+cd "$(dirname "$0")/.."
+mvn -B -q test "$@"
+ls -1 output
diff --git a/share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java b/share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java
new file mode 100644
index 0000000..a2eb781
--- /dev/null
+++ b/share-groups/src/main/java/com/ankurm/sharegroups/PaymentListeners.java
@@ -0,0 +1,54 @@
+package com.ankurm.sharegroups;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.support.ShareAcknowledgment;
+import org.springframework.stereotype.Component;
+
+/** Two listeners on share containers. They only record what they saw; the tests read the lists. */
+@Component
+public class PaymentListeners {
+
+ public final List implicitSeen = new CopyOnWriteArrayList<>();
+
+ public final List manualSeen = new CopyOnWriteArrayList<>();
+
+ /** Implicit mode: returning normally acknowledges (ACCEPT); throwing makes the container reject or release. */
+ @KafkaListener(id = "implicit", topics = "${share.topic.implicit}", groupId = "${share.group.implicit}",
+ containerFactory = "implicitShareFactory")
+ void implicit(ConsumerRecord record) throws InterruptedException {
+ Thread.sleep(20);
+ implicitSeen.add(Thread.currentThread().getName() + " <- " + record.value());
+ }
+
+ /** Manual mode: the listener decides per record. "bad" is rejected, "again" is released once, the rest accepted. */
+ @KafkaListener(id = "manual", topics = "${share.topic.manual}", groupId = "${share.group.manual}",
+ containerFactory = "explicitShareFactory")
+ void manual(ConsumerRecord record, ShareAcknowledgment ack) {
+ int attempt = record.deliveryCount().orElse((short) 1);
+ switch (record.value()) {
+ case "bad" -> {
+ manualSeen.add("bad delivery #" + attempt + " -> reject");
+ ack.reject();
+ }
+ case "again" -> {
+ if (attempt == 1) {
+ manualSeen.add("again delivery #1 -> release");
+ ack.release();
+ }
+ else {
+ manualSeen.add("again delivery #" + attempt + " -> acknowledge");
+ ack.acknowledge();
+ }
+ }
+ default -> {
+ manualSeen.add(record.value() + " delivery #" + attempt + " -> acknowledge");
+ ack.acknowledge();
+ }
+ }
+ }
+}
diff --git a/share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java b/share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java
new file mode 100644
index 0000000..f9efef3
--- /dev/null
+++ b/share-groups/src/main/java/com/ankurm/sharegroups/ShareConfig.java
@@ -0,0 +1,53 @@
+package com.ankurm.sharegroups;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.EnableKafka;
+import org.springframework.kafka.config.ShareKafkaListenerContainerFactory;
+import org.springframework.kafka.core.DefaultShareConsumerFactory;
+import org.springframework.kafka.core.ShareConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties.ShareAckMode;
+
+/**
+ * Spring Boot 4.1.1 does not auto-configure share consumers, so the factory and the listener container
+ * factories are plain beans. Two container factories: implicit (the container acknowledges for you) and
+ * explicit (the listener must call ShareAcknowledgment).
+ */
+@Configuration
+@EnableKafka
+public class ShareConfig {
+
+ @Bean
+ ShareConsumerFactory shareConsumerFactory(@Value("${broker:localhost:9092}") String bootstrap) {
+ Map 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 implicitShareFactory(ShareConsumerFactory factory) {
+ ShareKafkaListenerContainerFactory f = new ShareKafkaListenerContainerFactory<>(factory);
+ f.setConcurrency(4);
+ f.getContainerProperties().setShareAckMode(ShareAckMode.IMPLICIT);
+ return f;
+ }
+
+ @Bean
+ ShareKafkaListenerContainerFactory explicitShareFactory(ShareConsumerFactory factory) {
+ ShareKafkaListenerContainerFactory f = new ShareKafkaListenerContainerFactory<>(factory);
+ f.setConcurrency(2);
+ f.getContainerProperties().setShareAckMode(ShareAckMode.MANUAL);
+ return f;
+ }
+}
diff --git a/share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java b/share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java
new file mode 100644
index 0000000..75ed51d
--- /dev/null
+++ b/share-groups/src/main/java/com/ankurm/sharegroups/ShareGroupsApplication.java
@@ -0,0 +1,9 @@
+package com.ankurm.sharegroups;
+
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+/** Marker application class so the Spring tests have something to bootstrap. Nothing here starts a broker. */
+@SpringBootApplication
+public class ShareGroupsApplication {
+
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java
new file mode 100644
index 0000000..d02b8e2
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/AckTypesTest.java
@@ -0,0 +1,75 @@
+package com.ankurm.sharegroups;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** What ACCEPT, RELEASE and REJECT do to a record, and what happens to a record that is released forever. */
+class AckTypesTest {
+
+ @Test
+ void acknowledgementTypes() throws Exception {
+ Transcript t = new Transcript("02-ack-types.txt");
+ String topic = Broker.freshTopic("acks", 1);
+ String group = topic + "-g";
+ Broker.shareGroupFromEarliest(group);
+ Broker.produce(topic, List.of(new String[] { "k", "ok" }, new String[] { "k", "retry-once" },
+ new String[] { "k", "reject-me" }, new String[] { "k", "always-release" }));
+
+ Properties p = new Properties();
+ p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP);
+ p.put(ConsumerConfig.GROUP_ID_CONFIG, group);
+ p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put("share.acknowledgement.mode", "explicit");
+
+ List deliveries = new ArrayList<>();
+ long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(40);
+ long lastRecord = System.nanoTime();
+ try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) {
+ c.subscribe(List.of(topic));
+ while (System.nanoTime() < end && System.nanoTime() - lastRecord < TimeUnit.SECONDS.toNanos(15)) {
+ for (ConsumerRecord r : c.poll(Duration.ofMillis(500))) {
+ lastRecord = System.nanoTime();
+ int attempt = r.deliveryCount().orElse((short) -1);
+ AcknowledgeType type = switch (r.value()) {
+ case "ok" -> AcknowledgeType.ACCEPT;
+ case "retry-once" -> attempt == 1 ? AcknowledgeType.RELEASE : AcknowledgeType.ACCEPT;
+ case "reject-me" -> AcknowledgeType.REJECT;
+ default -> AcknowledgeType.RELEASE;
+ };
+ c.acknowledge(r, type);
+ String line = String.format("%-15s delivery #%d -> %s", r.value(), attempt, type);
+ deliveries.add(line);
+ t.line("%s", line);
+ }
+ c.commitSync();
+ }
+ }
+ t.line("deliveries of ok = %d", count(deliveries, "ok "));
+ t.line("deliveries of retry-once = %d", count(deliveries, "retry-once"));
+ t.line("deliveries of reject-me = %d", count(deliveries, "reject-me"));
+ t.line("deliveries of always-release = %d", count(deliveries, "always-release"));
+ t.write();
+
+ assertThat(count(deliveries, "ok ")).isEqualTo(1);
+ assertThat(count(deliveries, "retry-once")).isEqualTo(2);
+ assertThat(count(deliveries, "reject-me")).isEqualTo(1);
+ assertThat(count(deliveries, "always-release")).isGreaterThan(1);
+ }
+
+ private long count(List lines, String prefix) {
+ return lines.stream().filter(l -> l.startsWith(prefix)).count();
+ }
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/Broker.java b/share-groups/src/test/java/com/ankurm/sharegroups/Broker.java
new file mode 100644
index 0000000..204d7d4
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/Broker.java
@@ -0,0 +1,67 @@
+package com.ankurm.sharegroups;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.AlterConfigOp;
+import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.config.ConfigResource;
+import org.apache.kafka.common.serialization.StringSerializer;
+
+/**
+ * Helpers for the real broker. The tests do not start a broker: run scripts/broker-up.sh first
+ * (Kafka 4.3.1, KRaft, localhost:9092). Override with -Dbroker=host:port.
+ */
+final class Broker {
+
+ static final String BOOTSTRAP = System.getProperty("broker", "localhost:9092");
+
+ private Broker() {
+ }
+
+ static Properties admin() {
+ Properties p = new Properties();
+ p.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
+ return p;
+ }
+
+ /** A fresh topic per test run (unique suffix), so reruns never see old records. */
+ static String freshTopic(String base, int partitions) throws Exception {
+ String name = base + "-" + System.currentTimeMillis();
+ try (AdminClient admin = AdminClient.create(admin())) {
+ admin.createTopics(List.of(new NewTopic(name, partitions, (short) 1))).all().get(30, TimeUnit.SECONDS);
+ }
+ return name;
+ }
+
+ /** Share groups start at the latest offset by default; flip a group to earliest before it first subscribes. */
+ static void shareGroupFromEarliest(String group) throws Exception {
+ ConfigResource resource = new ConfigResource(ConfigResource.Type.GROUP, group);
+ try (AdminClient admin = AdminClient.create(admin())) {
+ admin.incrementalAlterConfigs(Map.of(resource,
+ List.of(new AlterConfigOp(new ConfigEntry("share.auto.offset.reset", "earliest"), AlterConfigOp.OpType.SET))))
+ .all().get(30, TimeUnit.SECONDS);
+ }
+ }
+
+ static void produce(String topic, List keyValues) {
+ Properties p = new Properties();
+ p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP);
+ p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
+ try (KafkaProducer producer = new KafkaProducer<>(p)) {
+ for (String[] kv : keyValues) {
+ producer.send(new ProducerRecord<>(topic, kv[0], kv[1]));
+ }
+ producer.flush();
+ }
+ }
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java
new file mode 100644
index 0000000..6d72561
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/OffsetResetTest.java
@@ -0,0 +1,60 @@
+package com.ankurm.sharegroups;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** The first surprise: a brand-new share group starts at the end of the topic, not the beginning. */
+class OffsetResetTest {
+
+ @Test
+ void newShareGroupSeesNothingUntilItIsToldToStartEarly() throws Exception {
+ Transcript t = new Transcript("03-share-group-starts-at-latest.txt");
+ String topic = Broker.freshTopic("reset", 1);
+ Broker.produce(topic, List.of(new String[] { "k", "a" }, new String[] { "k", "b" }, new String[] { "k", "c" }));
+ t.line("produced 3 records before any group existed");
+
+ int byDefault = drain(topic, topic + "-default", t);
+ t.line("new group, default share.auto.offset.reset -> received %d records", byDefault);
+
+ String early = topic + "-early";
+ Broker.shareGroupFromEarliest(early);
+ int withEarliest = drain(topic, early, t);
+ t.line("new group, share.auto.offset.reset=earliest -> received %d records", withEarliest);
+ t.write();
+
+ assertThat(byDefault).isZero();
+ assertThat(withEarliest).isEqualTo(3);
+ }
+
+ private int drain(String topic, String group, Transcript t) {
+ Properties p = new Properties();
+ p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP);
+ p.put(ConsumerConfig.GROUP_ID_CONFIG, group);
+ p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put("share.acknowledgement.mode", "explicit");
+ int received = 0;
+ try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) {
+ c.subscribe(List.of(topic));
+ long end = System.currentTimeMillis() + 12_000;
+ while (System.currentTimeMillis() < end) {
+ for (ConsumerRecord r : c.poll(Duration.ofMillis(500))) {
+ c.acknowledge(r, AcknowledgeType.ACCEPT);
+ received++;
+ }
+ c.commitSync();
+ }
+ }
+ return received;
+ }
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java
new file mode 100644
index 0000000..3d9eea7
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/OrderingTest.java
@@ -0,0 +1,92 @@
+package com.ankurm.sharegroups;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** 40 records, one key, one partition, three consumers with uneven speed: arrival order is not the produce order. */
+class OrderingTest {
+
+ @Test
+ void sameKeyDoesNotMeanSameOrder() throws Exception {
+ Transcript t = new Transcript("04-no-ordering-guarantee.txt");
+ String topic = Broker.freshTopic("order", 1);
+ String group = topic + "-g";
+ Broker.shareGroupFromEarliest(group);
+ List records = new ArrayList<>();
+ for (int i = 0; i < 40; i++) {
+ records.add(new String[] { "same-key", String.valueOf(i) });
+ }
+
+ List arrival = Collections.synchronizedList(new ArrayList<>());
+ AtomicInteger done = new AtomicInteger();
+ ExecutorService pool = Executors.newFixedThreadPool(3);
+ int[] workMillis = { 5, 40, 90 };
+ for (int id = 0; id < 3; id++) {
+ int work = workMillis[id];
+ pool.submit(() -> {
+ Properties p = new Properties();
+ p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP);
+ p.put(ConsumerConfig.GROUP_ID_CONFIG, group);
+ p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "4");
+ p.put("share.acknowledgement.mode", "explicit");
+ p.put("share.acquire.mode", "record_limit");
+ try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) {
+ c.subscribe(List.of(topic));
+ while (done.get() < 40 && !Thread.currentThread().isInterrupted()) {
+ for (ConsumerRecord r : c.poll(Duration.ofMillis(300))) {
+ Thread.sleep(work);
+ arrival.add(Integer.valueOf(r.value()));
+ c.acknowledge(r, AcknowledgeType.ACCEPT);
+ done.incrementAndGet();
+ }
+ c.commitSync();
+ }
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ }
+ Thread.sleep(6000); // let all three consumers join the group before any record exists
+ Broker.produce(topic, records);
+ long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
+ while (done.get() < 40 && System.nanoTime() < end) {
+ Thread.sleep(100);
+ }
+ pool.shutdownNow();
+ pool.awaitTermination(15, TimeUnit.SECONDS);
+
+ List snapshot = new ArrayList<>(arrival);
+ long inversions = 0;
+ for (int i = 1; i < snapshot.size(); i++) {
+ if (snapshot.get(i) < snapshot.get(i - 1)) {
+ inversions++;
+ }
+ }
+ t.line("produced : 0..39 in order, one key, one partition");
+ t.line("processed : %s", snapshot);
+ t.line("adjacent pairs that arrived out of order = %d", inversions);
+ t.write();
+
+ assertThat(snapshot).hasSize(40);
+ assertThat(inversions).isGreaterThan(0);
+ }
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java
new file mode 100644
index 0000000..9047c99
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/SpringShareListenerTest.java
@@ -0,0 +1,71 @@
+package com.ankurm.sharegroups;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/** The same ideas through Spring Kafka 4.1.1's share containers instead of the raw client. */
+@SpringBootTest(classes = { ShareGroupsApplication.class })
+class SpringShareListenerTest {
+
+ static String implicitTopic;
+ static String manualTopic;
+
+ @DynamicPropertySource
+ static void props(DynamicPropertyRegistry r) throws Exception {
+ implicitTopic = Broker.freshTopic("spring-implicit", 1);
+ manualTopic = Broker.freshTopic("spring-manual", 1);
+ Broker.shareGroupFromEarliest(implicitTopic + "-g");
+ Broker.shareGroupFromEarliest(manualTopic + "-g");
+ r.add("share.topic.implicit", () -> implicitTopic);
+ r.add("share.group.implicit", () -> implicitTopic + "-g");
+ r.add("share.topic.manual", () -> manualTopic);
+ r.add("share.group.manual", () -> manualTopic + "-g");
+ r.add("spring.kafka.listener.auto-startup", () -> "true");
+ }
+
+ @Autowired
+ PaymentListeners listeners;
+
+ @Test
+ void implicitAndManualAcknowledgement() throws Exception {
+ Transcript t = new Transcript("05-spring-share-listeners.txt");
+
+ List batch = new ArrayList<>();
+ for (int i = 0; i < 24; i++) {
+ batch.add(new String[] { "k" + i, "p" + i });
+ }
+ Broker.produce(implicitTopic, batch);
+ await().atMost(Duration.ofSeconds(60)).until(() -> listeners.implicitSeen.size() >= 24);
+
+ Set threads = new TreeSet<>();
+ listeners.implicitSeen.forEach(s -> threads.add(s.substring(0, s.indexOf(" <- "))));
+ t.line("== implicit mode, concurrency 4, 24 records on one topic partition");
+ t.line("records handled = %d", listeners.implicitSeen.size());
+ t.line("distinct listener threads that handled records = %d", threads.size());
+ threads.forEach(n -> t.line(" %s", n));
+
+ Broker.produce(manualTopic, List.of(new String[] { "k", "ok" }, new String[] { "k", "again" }, new String[] { "k", "bad" }));
+ await().atMost(Duration.ofSeconds(60)).until(() -> listeners.manualSeen.size() >= 4);
+ Thread.sleep(8000);
+ t.line("== manual mode (ShareAckMode.MANUAL), listener takes a ShareAcknowledgment");
+ listeners.manualSeen.stream().sorted().forEach(s -> t.line("%s", s));
+ t.write();
+
+ assertThat(listeners.implicitSeen).hasSize(24);
+ assertThat(listeners.manualSeen).anyMatch(s -> s.startsWith("again delivery #2"));
+ assertThat(listeners.manualSeen.stream().filter(s -> s.startsWith("bad")).count()).isEqualTo(1);
+ }
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java b/share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java
new file mode 100644
index 0000000..642518b
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/Transcript.java
@@ -0,0 +1,30 @@
+package com.ankurm.sharegroups;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Collects the lines a test prints and writes them to output/NN-name.txt, so the article quotes real output. */
+final class Transcript {
+
+ private final String file;
+ private final List lines = new ArrayList<>();
+
+ Transcript(String file) {
+ this.file = file;
+ }
+
+ synchronized void line(String format, Object... args) {
+ String l = String.format(format, args);
+ lines.add(l);
+ System.out.println(l);
+ }
+
+ void write() throws IOException {
+ Path dir = Path.of("output");
+ Files.createDirectories(dir);
+ Files.write(dir.resolve(file), lines);
+ }
+}
diff --git a/share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java b/share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java
new file mode 100644
index 0000000..9b8dc77
--- /dev/null
+++ b/share-groups/src/test/java/com/ankurm/sharegroups/WorkSpreadTest.java
@@ -0,0 +1,146 @@
+package com.ankurm.sharegroups;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.KafkaShareConsumer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Six consumers, a two-partition topic, 60 records that each take 50 ms to process.
+ * A consumer group can keep at most two of the six busy; a share group can keep more busy, but only when
+ * share.acquire.mode lets max.poll.records bound what one consumer takes (see the default-mode result).
+ */
+class WorkSpreadTest {
+
+ static final int CONSUMERS = 6;
+ static final int RECORDS = 60;
+ static final int PARTITIONS = 2;
+
+ record Result(ConcurrentHashMap perConsumer, long millis) {
+ }
+
+ @Test
+ void consumerGroupVersusShareGroup() throws Exception {
+ Transcript t = new Transcript("01-work-spread.txt");
+ t.line("topic partitions = %d, consumers = %d, records = %d, work per record = 50 ms", PARTITIONS, CONSUMERS, RECORDS);
+
+ Result classic = run("classic", t);
+ Result shareDefault = run("share-default", t);
+ Result share = run("share-record-limit", t);
+
+ long activeClassic = classic.perConsumer().values().stream().filter(c -> c.get() > 0).count();
+ long activeShare = share.perConsumer().values().stream().filter(c -> c.get() > 0).count();
+ long activeDefault = shareDefault.perConsumer().values().stream().filter(c -> c.get() > 0).count();
+ t.line("consumer group : consumers that received records = %d", activeClassic);
+ t.line("share group, default share.acquire.mode: consumers that received records = %d", activeDefault);
+ t.line("share group, acquire.mode=record_limit : consumers that received records = %d", activeShare);
+ t.line("(elapsed times vary from run to run; the shape, not the milliseconds, is the result)");
+ t.write();
+
+ assertThat(activeClassic).isEqualTo(PARTITIONS);
+ assertThat(activeShare).isGreaterThan(PARTITIONS);
+ assertThat(activeDefault).isGreaterThanOrEqualTo(1);
+ }
+
+ private Result run(String variant, Transcript t) throws Exception {
+ boolean share = variant.startsWith("share");
+ String topic = Broker.freshTopic("spread-" + variant, PARTITIONS);
+ String group = topic + "-g";
+ if (share) {
+ Broker.shareGroupFromEarliest(group);
+ }
+ ConcurrentHashMap counts = new ConcurrentHashMap<>();
+ AtomicInteger total = new AtomicInteger();
+ ExecutorService pool = Executors.newFixedThreadPool(CONSUMERS);
+ for (int i = 0; i < CONSUMERS; i++) {
+ int id = i;
+ counts.put(id, new AtomicInteger());
+ pool.submit(() -> consume(variant, topic, group, id, counts, total));
+ }
+ Thread.sleep(6000); // let every consumer join before the records arrive
+ List records = new ArrayList<>();
+ for (int i = 0; i < RECORDS; i++) {
+ records.add(new String[] { "key-" + i, "r" + i });
+ }
+ long start = System.nanoTime();
+ Broker.produce(topic, records);
+ while (total.get() < RECORDS && System.nanoTime() - start < TimeUnit.SECONDS.toNanos(60)) {
+ Thread.sleep(50);
+ }
+ long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
+ pool.shutdownNow();
+ pool.awaitTermination(15, TimeUnit.SECONDS);
+
+ t.line("== %s", switch (variant) {
+ case "classic" -> "consumer group (6 KafkaConsumers, same group.id)";
+ case "share-default" -> "share group (6 ShareConsumers, default share.acquire.mode)";
+ default -> "share group (6 ShareConsumers, share.acquire.mode=record_limit)";
+ });
+ counts.forEach((id, c) -> t.line("consumer-%d processed %d records", id, c.get()));
+ t.line("total processed = %d in %d ms", total.get(), millis);
+ return new Result(counts, millis);
+ }
+
+ private void consume(String variant, String topic, String group, int id, ConcurrentHashMap counts,
+ AtomicInteger total) {
+ Properties p = new Properties();
+ p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Broker.BOOTSTRAP);
+ p.put(ConsumerConfig.GROUP_ID_CONFIG, group);
+ p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
+ p.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "3");
+ boolean share = variant.startsWith("share");
+ if (variant.equals("share-record-limit")) {
+ p.put("share.acquire.mode", "record_limit");
+ }
+ try {
+ if (share) {
+ p.put("share.acknowledgement.mode", "explicit");
+ try (KafkaShareConsumer c = new KafkaShareConsumer<>(p)) {
+ c.subscribe(List.of(topic));
+ while (!Thread.currentThread().isInterrupted()) {
+ for (ConsumerRecord r : c.poll(Duration.ofMillis(300))) {
+ Thread.sleep(50);
+ c.acknowledge(r, AcknowledgeType.ACCEPT);
+ counts.get(id).incrementAndGet();
+ total.incrementAndGet();
+ }
+ c.commitSync();
+ }
+ }
+ }
+ else {
+ p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+ try (KafkaConsumer c = new KafkaConsumer<>(p)) {
+ c.subscribe(List.of(topic));
+ while (!Thread.currentThread().isInterrupted()) {
+ for (ConsumerRecord r : c.poll(Duration.ofMillis(300))) {
+ Thread.sleep(50);
+ counts.get(id).incrementAndGet();
+ total.incrementAndGet();
+ }
+ c.commitSync();
+ }
+ }
+ }
+ }
+ catch (InterruptedException | org.apache.kafka.common.errors.InterruptException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+}
diff --git a/share-groups/src/test/resources/logback-test.xml b/share-groups/src/test/resources/logback-test.xml
new file mode 100644
index 0000000..fab7eaa
--- /dev/null
+++ b/share-groups/src/test/resources/logback-test.xml
@@ -0,0 +1,8 @@
+
+
+ %d{HH:mm:ss.SSS} %-5level %logger{20} - %msg%n
+
+
+
+
+