Add concurrency support to ShareKafkaMessageListenerContainer

  Share consumers (KIP-932) enable record-level load balancing where
  multiple consumers can cooperatively process from the same partitions.
  Unlike traditional consumer groups with exclusive partition ownership,
  share groups distribute work at the broker level via the share group
  coordinator.

  This commit adds native concurrency support to the existing
  `ShareKafkaMessageListenerContainer` rather than creating a separate
  `ConcurrentShareKafkaMessageListenerContainer`. This design choice
  avoids the parent/child container complexity that exists in the regular
  consumer model, since share consumers fundamentally operate differently:

  - Work distribution happens at the broker level, not at the Spring layer
  - Multiple threads simply provide more capacity for the broker to
    distribute records across
  - No partition ownership model to coordinate between child containers

  This approach provides:

  - Simpler architecture with a single container managing multiple threads
  - No parent/child context propagation concerns
  - Better alignment with share consumer semantics (record-level vs
    partition-level distribution)
  - Increased throughput for high-volume workloads
  - Better resource utilization across consumer threads

  Users can configure concurrency at three levels:
  1. Per-listener via `@KafkaListener(concurrency = N)`
  2. Factory-level default via `factory.setConcurrency(N)`
  3. Programmatically via `container.setConcurrency(N)`

  The feature works seamlessly with both implicit (auto-acknowledge) and
  explicit (manual acknowledge/release/reject) acknowledgment modes, with
  each consumer thread independently managing its own acknowledgments.

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

Author: sobychacko

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

@@ -107,7 +107,7 @@ factory.addListener(new ShareConsumerFactory.Listener<String, String>() {
 [[share-kafka-message-listener-container]]
 === ShareKafkaMessageListenerContainer
 
-The `ShareKafkaMessageListenerContainer` provides a simple, single-threaded container for share consumers:
+The `ShareKafkaMessageListenerContainer` provides a container for share consumers with support for concurrent processing:
 
 [source,java]
 ----
@@ -151,6 +151,145 @@ Share consumers do not support:
 * Manual offset management
 ====
 
+[[share-container-concurrency]]
+=== Concurrency
+
+The `ShareKafkaMessageListenerContainer` supports concurrent processing by creating multiple consumer threads within a single container.
+Each thread runs its own `ShareConsumer` instance that participates in the same share group.
+
+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.
+
+==== Configuring Concurrency Programmatically
+
+[source,java]
+----
+@Bean
+public ShareKafkaMessageListenerContainer<String, String> concurrentContainer(
+        ShareConsumerFactory<String, String> shareConsumerFactory) {
+
+    ContainerProperties containerProps = new ContainerProperties("my-topic");
+    containerProps.setGroupId("my-share-group");
+
+    ShareKafkaMessageListenerContainer<String, String> container =
+        new ShareKafkaMessageListenerContainer<>(shareConsumerFactory, containerProps);
+
+    // Set concurrency to create 5 consumer threads
+    container.setConcurrency(5);
+
+    container.setupMessageListener(new MessageListener<String, String>() {
+        @Override
+        public void onMessage(ConsumerRecord<String, String> record) {
+            System.out.println("Received on " + Thread.currentThread().getName() + ": " + record.value());
+        }
+    });
+
+    return container;
+}
+----
+
+==== Configuring Concurrency via Factory
+
+You can set default concurrency at the factory level, which applies to all containers created by that factory:
+
+[source,java]
+----
+@Bean
+public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
+        ShareConsumerFactory<String, String> shareConsumerFactory) {
+
+    ShareKafkaListenerContainerFactory<String, String> factory =
+        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
+
+    // Set default concurrency for all containers created by this factory
+    factory.setConcurrency(3);
+
+    return factory;
+}
+----
+
+==== Per-Listener Concurrency
+
+The concurrency setting can be overridden per listener using the `concurrency` attribute:
+
+[source,java]
+----
+@Component
+public class ConcurrentShareListener {
+
+    @KafkaListener(
+        topics = "high-throughput-topic",
+        containerFactory = "shareKafkaListenerContainerFactory",
+        groupId = "my-share-group",
+        concurrency = "10"  // Override factory default
+    )
+    public void listen(ConsumerRecord<String, String> record) {
+        // This listener will use 10 consumer threads
+        System.out.println("Processing: " + record.value());
+    }
+}
+----
+
+==== Concurrency Considerations
+
+* **Thread Safety**: Each consumer thread has its own `ShareConsumer` instance and manages its own acknowledgments independently
+* **Client IDs**: Each consumer thread receives a unique client ID with a numeric suffix (e.g., `myContainer-0`, `myContainer-1`, etc.)
+* **Metrics**: Metrics from all consumer threads are aggregated and accessible via `container.metrics()`
+* **Lifecycle**: All consumer threads start and stop together as a unit
+* **Work Distribution**: The Kafka broker handles record distribution across all consumer instances in the share group
+* **Explicit Acknowledgment**: Each thread independently manages acknowledgments for its records; unacknowledged records in one thread don't block other threads
+
+==== Concurrency with Explicit Acknowledgment
+
+Concurrency works seamlessly with explicit acknowledgment mode.
+Each consumer thread independently tracks and acknowledges its own records:
+
+[source,java]
+----
+@KafkaListener(
+    topics = "order-queue",
+    containerFactory = "explicitShareKafkaListenerContainerFactory",
+    groupId = "order-processors",
+    concurrency = "5"
+)
+public void processOrder(ConsumerRecord<String, String> record, ShareAcknowledgment acknowledgment) {
+    try {
+        // Process the order
+        processOrderLogic(record.value());
+        acknowledgment.acknowledge(); // ACCEPT
+    }
+    catch (RetryableException e) {
+        acknowledgment.release(); // Will be redelivered
+    }
+    catch (Exception e) {
+        acknowledgment.reject(); // Permanent failure
+    }
+}
+----
+
+[NOTE]
+====
+**Work Distribution Behavior:**
+
+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:
+
+* The topic has a single partition
+* There's low message volume
+* The broker's distribution algorithm favors certain consumers
+
+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.
+
+This differs from traditional `ConcurrentMessageListenerContainer` where Spring explicitly distributes partitions across threads.
+
+When using concurrency with share consumers:
+
+* Each thread polls and processes records independently
+* Acknowledgment constraints apply per-thread (one thread's unacknowledged records don't block other threads)
+* Concurrency setting must be greater than 0 and cannot be changed while the container is running
+====
+
 [[share-annotation-driven-listeners]]
 == Annotation-Driven Listeners
 
@@ -520,8 +659,7 @@ Share consumers differ from regular consumers in several key ways:
 === Current Limitations
 
 * **In preview**: This feature is in preview mode and may change in future versions
-* **Single-Threaded**: Share consumer containers currently run in single-threaded mode
 * **No Message Converters**: Message converters are not yet supported for share consumers
 * **No Batch Listeners**: Batch processing is not supported with share consumers
-* **Poll Constraints**: In explicit acknowledgment mode, unacknowledged records block subsequent polls
+* **Poll Constraints**: In explicit acknowledgment mode, unacknowledged records block subsequent polls within each consumer thread
 

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

@@ -22,6 +22,7 @@
 
 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;
@@ -63,6 +64,8 @@ public class ShareKafkaListenerContainerFactory<K, V>
 
 	private int phase = 0;
 
+	private @Nullable Integer concurrency;
+
 	@SuppressWarnings("NullAway.Init")
 	private ApplicationEventPublisher applicationEventPublisher;
 
@@ -98,6 +101,22 @@ public void setPhase(int phase) {
 		this.phase = phase;
 	}
 
+	/**
+	 * Set the concurrency for containers created by this factory.
+	 * <p>
+	 * This specifies the number of consumer threads to create within each container.
+	 * Each thread creates its own {@link org.apache.kafka.clients.consumer.ShareConsumer}
+	 * instance and participates in the same share group. The Kafka broker distributes
+	 * records across all consumer instances, providing record-level load balancing.
+	 * <p>
+	 * This can be overridden per listener endpoint using the {@code concurrency}
+	 * attribute on {@code @KafkaListener}.
+	 * @param concurrency the number of consumer threads (must be greater than 0)
+	 */
+	public void setConcurrency(Integer concurrency) {
+		this.concurrency = concurrency;
+	}
+
 	@Override
 	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
 		this.applicationEventPublisher = applicationEventPublisher;
@@ -138,6 +157,15 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 		boolean explicitAck = determineExplicitAcknowledgment(properties);
 		properties.setExplicitShareAcknowledgment(explicitAck);
 
+		// Set concurrency - endpoint setting takes precedence over factory setting
+		Integer conc = endpoint.getConcurrency();
+		if (conc != null) {
+			instance.setConcurrency(conc);
+		}
+		else if (this.concurrency != null) {
+			instance.setConcurrency(this.concurrency);
+		}
+
 		instance.setAutoStartup(effectiveAutoStartup);
 		instance.setPhase(this.phase);
 		instance.setApplicationContext(this.applicationContext);