Add share-groups module: Kafka 4.3.1 share groups with Spring for Apache Kafka 4.1.1

Six consumers on two partitions under a consumer group, a share group with the default share.acquire.mode and a share group with record_limit; the three acknowledgement types and a record that is released until Kafka stops redelivering it; a new share group starting at latest; the ordering a share group gives up; and Spring Kafka's share containers in implicit and manual acknowledgement modes. Tests run against a real broker (scripts/broker-up.sh) and write output/. Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01Lm655hdgDo5xsSLNomgdsy
2026-10-13 02:43:36 +00:00
parent 4a5d3318bf
commit 5db56e70a0
23 changed files with 889 additions and 2 deletions
@@ -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
+output under `docs/output/` alongside numbered documentation chapters; `saga/` and `share-groups/` keep it
-under `output/` instead, with the equivalent depth in the article's own accordion sections.
+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 &mdash; choreography and orchestration &mdash; 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
@@ -0,0 +1,2 @@
 target/
 .kafka/
@@ -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.
@@ -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)
@@ -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
@@ -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
@@ -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
@@ -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
@@ -0,0 +1,52 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0">
  <modelVersion>4.0.0</modelVersion>
  <!-- Same convention as the rest of this repo: inherit spring-boot-starter-parent. Boot 4.1.1
       manages Spring Kafka 4.1.1 and kafka-clients 4.2.1. -->
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>4.1.1</version>
    <relativePath/>
  </parent>
  <groupId>com.ankurm</groupId>
  <artifactId>share-groups</artifactId>
  <version>1.0</version>
  <packaging>jar</packaging>
  <properties>
    <java.version>25</java.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.awaitility</groupId>
      <artifactId>awaitility</artifactId>
      <scope>test</scope>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
 </project>
@@ -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
@@ -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
@@ -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:<port> if it is not 9092.
 set -euo pipefail
 cd "$(dirname "$0")/.."
 mvn -B -q test "$@"
 ls -1 output
@@ -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<String> implicitSeen = new CopyOnWriteArrayList<>();
 	public final List<String> 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<String, String> 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<String, String> 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();
 			}
 		}
 	}
 }
@@ -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<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> implicitShareFactory(ShareConsumerFactory<String, String> factory) {
 		ShareKafkaListenerContainerFactory<String, String> f = new ShareKafkaListenerContainerFactory<>(factory);
 		f.setConcurrency(4);
 		f.getContainerProperties().setShareAckMode(ShareAckMode.IMPLICIT);
 		return f;
 	}
 	@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;
 	}
 }
@@ -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 {
 }
@@ -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<String> deliveries = new ArrayList<>();
 		long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(40);
 		long lastRecord = System.nanoTime();
 		try (KafkaShareConsumer<String, String> c = new KafkaShareConsumer<>(p)) {
 			c.subscribe(List.of(topic));
 			while (System.nanoTime() < end && System.nanoTime() - lastRecord < TimeUnit.SECONDS.toNanos(15)) {
 				for (ConsumerRecord<String, String> 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<String> lines, String prefix) {
 		return lines.stream().filter(l -> l.startsWith(prefix)).count();
 	}
 }
@@ -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<String[]> 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<String, String> producer = new KafkaProducer<>(p)) {
 			for (String[] kv : keyValues) {
 				producer.send(new ProducerRecord<>(topic, kv[0], kv[1]));
 			}
 			producer.flush();
 		}
 	}
 }
@@ -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<String, String> c = new KafkaShareConsumer<>(p)) {
 			c.subscribe(List.of(topic));
 			long end = System.currentTimeMillis() + 12_000;
 			while (System.currentTimeMillis() < end) {
 				for (ConsumerRecord<String, String> r : c.poll(Duration.ofMillis(500))) {
 					c.acknowledge(r, AcknowledgeType.ACCEPT);
 					received++;
 				}
 				c.commitSync();
 			}
 		}
 		return received;
 	}
 }
@@ -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<String[]> records = new ArrayList<>();
 		for (int i = 0; i < 40; i++) {
 			records.add(new String[] { "same-key", String.valueOf(i) });
 		}
 		List<Integer> 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<String, String> c = new KafkaShareConsumer<>(p)) {
 					c.subscribe(List.of(topic));
 					while (done.get() < 40 && !Thread.currentThread().isInterrupted()) {
 						for (ConsumerRecord<String, String> 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<Integer> 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);
 	}
 }
@@ -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<String[]> 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<String> 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);
 	}
 }
@@ -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<String> 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);
 	}
 }
@@ -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<Integer, AtomicInteger> 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<Integer, AtomicInteger> 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<String[]> 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<Integer, AtomicInteger> 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<String, String> c = new KafkaShareConsumer<>(p)) {
 					c.subscribe(List.of(topic));
 					while (!Thread.currentThread().isInterrupted()) {
 						for (ConsumerRecord<String, String> 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<String, String> c = new KafkaConsumer<>(p)) {
 					c.subscribe(List.of(topic));
 					while (!Thread.currentThread().isInterrupted()) {
 						for (ConsumerRecord<String, String> 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();
 		}
 	}
 }
@@ -0,0 +1,8 @@
 <configuration>
  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
    <encoder><pattern>%d{HH:mm:ss.SSS} %-5level %logger{20} - %msg%n</pattern></encoder>
  </appender>
  <logger name="org.apache.kafka" level="OFF"/>
  <logger name="org.springframework.kafka" level="WARN"/>
  <root level="WARN"><appender-ref ref="STDOUT"/></root>
 </configuration>