Address PR feedback on concurrency implementation

  - Use primitive int for concurrency in factory (consistent with phase field)
  - Remove unnecessary `getConcurrency()` getter (only used in trivial tests)
  - Use `HashMap` instead of `ConcurrentHashMap` in metrics() (already inside lock)
  - Use `CompletableFuture.allOf()` for cleaner shutdown coordination
  - Remove debug logging from tests (unnecessary noise in CI/CD)
  - Remove thread tracking from concurrency tests (over-complicates assertions)

  Clarify documentation based on KIP-932 specifications:

  - Add explicit note that concurrency is additive across application instances
  - Replace high-level distribution description with precise KIP-932 details
  - Document pull-based model, acquisition locks, and batch behavior
  - Explain `max.poll.records` as soft limit with complete batch preference
  - Set accurate expectations about broker-controlled record distribution

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/kafka-queues.adoc

@@ -160,6 +160,23 @@ Each thread runs its own `ShareConsumer` instance that participates in the same
 Unlike traditional consumer groups where concurrency involves partition distribution, share consumers leverage Kafka's record-level distribution at the broker.
 This means multiple consumer threads in the same container work together as part of the share group, with the Kafka broker distributing records across all consumer instances.
 
+[IMPORTANT]
+====
+**Concurrency is Additive Across Application Instances**
+
+From the share group's perspective, each `ShareConsumer` instance is an independent member, regardless of where it runs.
+Setting `concurrency=3` in a single container creates 3 share group members.
+If you run multiple application instances with the same share group ID, all their consumer threads combine into one pool.
+
+For example:
+* Application Instance 1: `concurrency=3` → 3 share group members
+* Application Instance 2: `concurrency=3` → 3 share group members
+* **Total**: 6 share group members available for the broker to distribute records to
+
+This means setting `concurrency=5` in a single container is operationally equivalent to running 5 separate application instances with `concurrency=1` each (all using the same `group.id`).
+The Kafka broker treats all consumer instances equally and distributes records across the entire pool.
+====
+
 ==== Configuring Concurrency Programmatically
 
 [source,java]
@@ -269,21 +286,31 @@ public void processOrder(ConsumerRecord<String, String> record, ShareAcknowledgm
 
 [NOTE]
 ====
-**Work Distribution Behavior:**
+**Record Acquisition and Distribution Behavior:**
+
+Share consumers use a pull-based model where each consumer thread calls `poll()` to fetch records from the broker.
+When a consumer polls, the broker's share-partition leader:
+
+* Selects records in "Available" state
+* Moves them to "Acquired" state with a time-limited acquisition lock (default 30 seconds, configurable via `group.share.record.lock.duration.ms`)
+* Prefers to return complete record batches for efficiency
+* Applies `max.poll.records` as a soft limit, meaning complete record batches will be acquired even if it exceeds this value
 
-With share consumers, record distribution is controlled by Kafka's share group coordinator at the broker level, not by Spring for Apache Kafka.
-The broker may assign all records to a single consumer thread at any given time, especially when:
+While records are acquired by one consumer, they are not available to other consumers.
+When the acquisition lock expires, unacknowledged records automatically return to "Available" state and can be delivered to another consumer.
 
-* The topic has a single partition
-* There's low message volume
-* The broker's distribution algorithm favors certain consumers
+The broker limits the number of records that can be acquired per partition using `group.share.partition.max.record.locks`.
+Once this limit is reached, subsequent polls temporarily return no records until locks expire.
 
-This is normal behavior. The key benefit of concurrency is having multiple consumer threads *available* to the share group coordinator for distribution.
-As message volume increases or over time, you should see distribution across multiple threads.
+**Implications for Concurrency:**
 
-This differs from traditional `ConcurrentMessageListenerContainer` where Spring explicitly distributes partitions across threads.
+* Each consumer thread independently polls and may acquire different numbers of records per poll
+* Record distribution across threads depends on polling timing and batch availability
+* Multiple threads increase the pool of consumers available to acquire records
+* With low message volume or single partitions, records may concentrate on fewer threads
+* For long-running workloads, distribution tends to be more even
 
-When using concurrency with share consumers:
+**Configuration:**
 
 * Each thread polls and processes records independently
 * Acknowledgment constraints apply per-thread (one thread's unacknowledged records don't block other threads)

spring-kafka/src/main/java/org/springframework/kafka/config/ShareKafkaListenerContainerFactory.java

@@ -22,7 +22,6 @@
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.internals.ShareAcknowledgementMode;
-import org.jspecify.annotations.Nullable;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -64,7 +63,7 @@ public class ShareKafkaListenerContainerFactory<K, V>
 
 	private int phase = 0;
 
-	private @Nullable Integer concurrency;
+	private int concurrency = 1;
 
 	@SuppressWarnings("NullAway.Init")
 	private ApplicationEventPublisher applicationEventPublisher;
@@ -113,7 +112,7 @@ public void setPhase(int phase) {
 	 * attribute on {@code @KafkaListener}.
 	 * @param concurrency the number of consumer threads (must be greater than 0)
 	 */
-	public void setConcurrency(Integer concurrency) {
+	public void setConcurrency(int concurrency) {
 		this.concurrency = concurrency;
 	}
 
@@ -162,7 +161,7 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 		if (conc != null) {
 			instance.setConcurrency(conc);
 		}
-		else if (this.concurrency != null) {
+		else {
 			instance.setConcurrency(this.concurrency);
 		}
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainer.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -134,14 +135,6 @@ public void setClientId(String clientId) {
 		this.clientId = clientId;
 	}
 
-	/**
-	 * Get the concurrency level (number of consumer threads).
-	 * @return the concurrency level
-	 */
-	public int getConcurrency() {
-		return this.concurrency;
-	}
-
 	/**
 	 * Set the level of concurrency. This will create the specified number of
 	 * consumer threads, each with its own {@link ShareConsumer} instance.
@@ -169,7 +162,7 @@ public boolean isInExpectedState() {
 			if (this.consumers.isEmpty()) {
 				return Collections.emptyMap();
 			}
-			Map<String, Map<MetricName, ? extends Metric>> allMetrics = new ConcurrentHashMap<>();
+			Map<String, Map<MetricName, ? extends Metric>> allMetrics = new HashMap<>();
 			for (ShareListenerConsumer consumer : this.consumers) {
 				Map<MetricName, ? extends Metric> consumerMetrics = consumer.consumer.metrics();
 				String consumerId = consumer.getClientId();
@@ -239,17 +232,13 @@ protected void doStop() {
 		// Wait for all consumer threads to complete
 		this.lifecycleLock.lock();
 		try {
-			for (CompletableFuture<Void> future : this.consumerFutures) {
-				try {
-					future.join(); // Wait for consumer to finish
-				}
-				catch (Exception e) {
-					this.logger.error(e, "Error waiting for consumer thread to stop");
-				}
-			}
+			CompletableFuture.allOf(this.consumerFutures.toArray(new CompletableFuture<?>[0])).join();
 			this.consumers.clear();
 			this.consumerFutures.clear();
 		}
+		catch (Exception e) {
+			this.logger.error(e, "Error waiting for consumer threads to stop");
+		}
 		finally {
 			this.lifecycleLock.unlock();
 		}

spring-kafka/src/test/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerIntegrationTests.java

@@ -32,7 +32,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
 
-import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.ConfigEntry;
@@ -84,8 +83,6 @@
 )
 class ShareKafkaMessageListenerContainerIntegrationTests {
 
-	private static final LogAccessor logger = new LogAccessor(LogFactory.getLog(ShareKafkaMessageListenerContainerIntegrationTests.class));
-
 	@Test
 	void integrationTestShareKafkaMessageListenerContainer(EmbeddedKafkaBroker broker) throws Exception {
 		final String topic = "share-listener-integration-test";
@@ -856,12 +853,9 @@ void shouldProcessRecordsWithMultipleConsumerThreads(EmbeddedKafkaBroker broker)
 
 		ContainerProperties containerProps = new ContainerProperties(topic);
 		CountDownLatch latch = new CountDownLatch(numRecords);
-		Map<String, AtomicInteger> threadCounts = new ConcurrentHashMap<>();
 		List<String> receivedValues = Collections.synchronizedList(new ArrayList<>());
 
 		containerProps.setMessageListener((MessageListener<String, String>) record -> {
-			String threadName = Thread.currentThread().getName();
-			threadCounts.computeIfAbsent(threadName, k -> new AtomicInteger()).incrementAndGet();
 			receivedValues.add(record.value());
 			latch.countDown();
 		});
@@ -871,34 +865,14 @@ void shouldProcessRecordsWithMultipleConsumerThreads(EmbeddedKafkaBroker broker)
 		container.setBeanName("concurrencyBasicTest");
 		container.setConcurrency(concurrency);
 
-		assertThat(container.getConcurrency()).isEqualTo(concurrency);
-
 		container.start();
 
 		try {
-			// Verify all records are processed
 			assertThat(latch.await(30, TimeUnit.SECONDS))
 					.as("All %d records should be processed", numRecords)
 					.isTrue();
 
-			// Log thread distribution for debugging
-			logger.info(() -> "Thread distribution: " + threadCounts);
-
-			// Verify all records received
 			assertThat(receivedValues).hasSize(numRecords);
-
-			// Share consumers with single partition may use only one consumer at a time
-			// So we verify: at least 1 thread used, at most concurrency threads
-			assertThat(threadCounts.size())
-					.as("At least one consumer thread should process records")
-					.isGreaterThanOrEqualTo(1)
-					.isLessThanOrEqualTo(concurrency);
-
-			// Verify work was done
-			int totalProcessed = threadCounts.values().stream()
-					.mapToInt(AtomicInteger::get)
-					.sum();
-			assertThat(totalProcessed).isEqualTo(numRecords);
 		}
 		finally {
 			container.stop();
@@ -953,8 +927,6 @@ void shouldAggregateMetricsFromMultipleConsumers(EmbeddedKafkaBroker broker) thr
 						.as("Client ID should contain bean name")
 						.contains("concurrencyMetricsTest");
 			}
-
-			logger.info(() -> "Client IDs from metrics: " + metrics.keySet());
 		}
 		finally {
 			container.stop();
@@ -983,13 +955,9 @@ void shouldHandleConcurrencyWithExplicitAcknowledgment(EmbeddedKafkaBroker broke
 		CountDownLatch latch = new CountDownLatch(numRecords);
 		AtomicInteger acceptCount = new AtomicInteger();
 		AtomicInteger rejectCount = new AtomicInteger();
-		Map<String, AtomicInteger> threadCounts = new ConcurrentHashMap<>();
 
 		containerProps.setMessageListener((AcknowledgingShareConsumerAwareMessageListener<String, String>) (
 				record, acknowledgment, consumer) -> {
-			String threadName = Thread.currentThread().getName();
-			threadCounts.computeIfAbsent(threadName, k -> new AtomicInteger()).incrementAndGet();
-
 			// Reject every 5th record, accept others
 			int recordNum = Integer.parseInt(record.value().substring(5)); // "value0" -> 0
 			if (recordNum % 5 == 0) {
@@ -1014,16 +982,6 @@ void shouldHandleConcurrencyWithExplicitAcknowledgment(EmbeddedKafkaBroker broke
 					.as("All records should be processed with explicit acknowledgment")
 					.isTrue();
 
-			logger.info(() -> "Thread distribution with explicit ack: " + threadCounts);
-			logger.info(() -> "Accept count: " + acceptCount.get() + ", Reject count: " + rejectCount.get());
-
-			// Verify at least one thread processed records
-			assertThat(threadCounts.size())
-					.as("At least one thread should process records in explicit mode")
-					.isGreaterThanOrEqualTo(1)
-					.isLessThanOrEqualTo(concurrency);
-
-			// Verify acknowledgments were processed correctly
 			assertThat(acceptCount.get() + rejectCount.get())
 					.as("Total acknowledgments should equal number of records")
 					.isEqualTo(numRecords);
@@ -1084,7 +1042,6 @@ void shouldStopAllConsumerThreadsGracefully(EmbeddedKafkaBroker broker) throws E
 				.untilAsserted(() -> assertThat(processedCount.get()).isGreaterThan(0));
 
 		int processedBeforeStop = processedCount.get();
-		logger.info(() -> "Processed " + processedBeforeStop + " records before stop");
 
 		// Stop the container
 		container.stop();
@@ -1112,8 +1069,6 @@ void shouldStopAllConsumerThreadsGracefully(EmbeddedKafkaBroker broker) throws E
 				.atMost(10, TimeUnit.SECONDS)
 				.untilAsserted(() -> assertThat(processedCount.get()).isGreaterThan(processedBeforeStop));
 
-		logger.info(() -> "Processed " + processedCount.get() + " records total after restart");
-
 		// Final stop
 		container.stop();
 		assertThat(container.isRunning()).isFalse();

spring-kafka/src/test/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerUnitTests.java

@@ -89,20 +89,6 @@ void shouldFailWhenExplicitModeUsedWithNonAcknowledgingListener() {
 				.withMessageContaining("Explicit acknowledgment mode requires an AcknowledgingShareConsumerAwareMessageListener");
 	}
 
-	@Test
-	void shouldSetConcurrencyCorrectly() {
-		ContainerProperties containerProperties = new ContainerProperties("test-topic");
-		containerProperties.setMessageListener(messageListener);
-
-		ShareKafkaMessageListenerContainer<String, String> container =
-				new ShareKafkaMessageListenerContainer<>(shareConsumerFactory, containerProperties);
-
-		assertThat(container.getConcurrency()).isEqualTo(1); // Default is 1
-
-		container.setConcurrency(5);
-		assertThat(container.getConcurrency()).isEqualTo(5);
-	}
-
 	@Test
 	void shouldRejectInvalidConcurrency() {
 		ContainerProperties containerProperties = new ContainerProperties("test-topic");