Add KafkaListener support for shared consumer containers

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

@@ -0,0 +1,214 @@
+/*
+ * Copyright 2025-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.regex.Pattern;
+
+import org.apache.commons.logging.LogFactory;
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.kafka.core.ShareConsumerFactory;
+import org.springframework.kafka.listener.AbstractShareKafkaMessageListenerContainer;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.support.JavaUtils;
+import org.springframework.kafka.support.TopicPartitionOffset;
+
+/**
+ * Base {@link KafkaListenerContainerFactory} for creating containers that use Kafka's share consumer model.
+ * <p>
+ * This abstract factory provides common configuration and lifecycle management for share consumer containers.
+ * It handles the creation of containers based on endpoints, topics, or patterns, and applies common
+ * configuration properties to the created containers.
+ * <p>
+ * The share consumer model enables cooperative rebalancing, allowing consumers to maintain ownership of
+ * some partitions while relinquishing others during rebalances, which can reduce disruption compared to
+ * the classic consumer model.
+ *
+ * @param <C> the container type
+ * @param <K> the key type
+ * @param <V> the value type
+ *
+ * @author Soby Chacko
+ * @since 4.0
+ */
+public abstract class AbstractShareKafkaListenerContainerFactory<C extends AbstractShareKafkaMessageListenerContainer<K, V>, K, V>
+		implements KafkaListenerContainerFactory<C>, ApplicationEventPublisherAware, ApplicationContextAware {
+
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));
+
+	private final ContainerProperties containerProperties = new ContainerProperties((Pattern) null);
+
+	private @Nullable ShareConsumerFactory<? super K, ? super V> shareConsumerFactory;
+
+	private @Nullable Boolean autoStartup;
+
+	private @Nullable Integer phase;
+
+	private @Nullable ApplicationEventPublisher applicationEventPublisher;
+
+	private @Nullable ApplicationContext applicationContext;
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) {
+		this.applicationContext = applicationContext;
+	}
+
+	/**
+	 * Set the share consumer factory to use for creating containers.
+	 * @param shareConsumerFactory the share consumer factory
+	 */
+	public void setShareConsumerFactory(ShareConsumerFactory<? super K, ? super V> shareConsumerFactory) {
+		this.shareConsumerFactory = shareConsumerFactory;
+	}
+
+	/**
+	 * Get the share consumer factory.
+	 * @return the share consumer factory
+	 */
+	public @Nullable ShareConsumerFactory<? super K, ? super V> getShareConsumerFactory() {
+		return this.shareConsumerFactory;
+	}
+
+	/**
+	 * Set whether containers created by this factory should auto-start.
+	 * @param autoStartup true to auto-start
+	 */
+	public void setAutoStartup(Boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	/**
+	 * Set the phase in which containers created by this factory should start and stop.
+	 * @param phase the phase
+	 */
+	public void setPhase(Integer phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+		this.applicationEventPublisher = applicationEventPublisher;
+	}
+
+	/**
+	 * Get the container properties.
+	 * @return the container properties
+	 */
+	public ContainerProperties getContainerProperties() {
+		return this.containerProperties;
+	}
+
+	@Override
+	public C createListenerContainer(KafkaListenerEndpoint endpoint) {
+		C instance = createContainerInstance(endpoint);
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
+		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
+			configureEndpoint((AbstractKafkaListenerEndpoint<K, V>) endpoint);
+		}
+		endpoint.setupListenerContainer(instance, null); // No message converter for MVP
+		initializeContainer(instance, endpoint);
+		return instance;
+	}
+
+	private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> endpoint) {
+		// Minimal configuration; can add more properties later
+	}
+
+	/**
+	 * Initialize the provided container with common configuration properties.
+	 * @param instance the container instance
+	 * @param endpoint the endpoint
+	 */
+	protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
+		ContainerProperties properties = instance.getContainerProperties();
+		if (this.containerProperties.getAckCount() > 0) {
+			properties.setAckCount(this.containerProperties.getAckCount());
+		}
+		if (this.containerProperties.getAckTime() > 0) {
+			properties.setAckTime(this.containerProperties.getAckTime());
+		}
+		if (endpoint.getAutoStartup() != null) {
+			instance.setAutoStartup(endpoint.getAutoStartup());
+		}
+		else if (this.autoStartup != null) {
+			instance.setAutoStartup(this.autoStartup);
+		}
+		if (this.phase != null) {
+			instance.setPhase(this.phase);
+		}
+		if (this.applicationContext != null) {
+			instance.setApplicationContext(this.applicationContext);
+		}
+		if (this.applicationEventPublisher != null) {
+			instance.setApplicationEventPublisher(this.applicationEventPublisher);
+		}
+		if (endpoint.getGroupId() != null) {
+			instance.getContainerProperties().setGroupId(endpoint.getGroupId());
+		}
+		if (endpoint.getClientIdPrefix() != null) {
+			instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
+		}
+		if (endpoint.getConsumerProperties() != null) {
+			instance.getContainerProperties().setKafkaConsumerProperties(endpoint.getConsumerProperties());
+		}
+	}
+
+	@Override
+	public C createContainer(TopicPartitionOffset... topicPartitions) {
+		return createContainerInstance(new KafkaListenerEndpointAdapter() {
+			@Override
+			public TopicPartitionOffset[] getTopicPartitionsToAssign() {
+				return Arrays.copyOf(topicPartitions, topicPartitions.length);
+			}
+		});
+	}
+
+	@Override
+	public C createContainer(String... topics) {
+		return createContainerInstance(new KafkaListenerEndpointAdapter() {
+			@Override
+			public Collection<String> getTopics() {
+				return Arrays.asList(topics);
+			}
+		});
+	}
+
+	@Override
+	public C createContainer(Pattern topicPattern) {
+		return createContainerInstance(new KafkaListenerEndpointAdapter() {
+			@Override
+			public Pattern getTopicPattern() {
+				return topicPattern;
+			}
+		});
+	}
+
+	/**
+	 * Create a container instance for the provided endpoint.
+	 * @param endpoint the endpoint
+	 * @return the container instance
+	 */
+	protected abstract C createContainerInstance(KafkaListenerEndpoint endpoint);
+}

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

@@ -31,11 +31,13 @@
 import org.springframework.expression.BeanResolver;
 import org.springframework.kafka.listener.KafkaListenerErrorHandler;
 import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.listener.ShareKafkaMessageListenerContainer;
 import org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.BatchToRecordAdapter;
 import org.springframework.kafka.listener.adapter.HandlerAdapter;
 import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
+import org.springframework.kafka.listener.adapter.ShareRecordMessagingMessageListenerAdapter;
 import org.springframework.kafka.support.JavaUtils;
 import org.springframework.kafka.support.converter.BatchMessageConverter;
 import org.springframework.kafka.support.converter.MessageConverter;
@@ -175,7 +177,14 @@ protected MessagingMessageListenerAdapter<K, V> createMessageListener(MessageLis
 
 		Assert.state(this.messageHandlerMethodFactory != null,
 				"Could not create message listener - MessageHandlerMethodFactory not set");
-		MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);
+
+		final MessagingMessageListenerAdapter<K, V> messageListener;
+		if (container instanceof ShareKafkaMessageListenerContainer<?, ?>) {
+			messageListener = createShareMessageListenerInstance(messageConverter);
+		}
+		else {
+			messageListener = createMessageListenerInstance(messageConverter);
+		}
 		messageListener.setHandlerMethod(configureListenerAdapter(messageListener));
 		JavaUtils.INSTANCE
 			.acceptIfNotNull(getReplyTopic(), replyTopic -> {
@@ -206,6 +215,26 @@ protected HandlerAdapter configureListenerAdapter(MessagingMessageListenerAdapte
 	 * @param messageConverter the converter (may be null).
 	 * @return the {@link MessagingMessageListenerAdapter} instance.
 	 */
+	protected MessagingMessageListenerAdapter<K, V> createShareMessageListenerInstance(
+			@Nullable MessageConverter messageConverter) {
+
+		MessagingMessageListenerAdapter<K, V> listener;
+		ShareRecordMessagingMessageListenerAdapter<K, V> messageListener = new ShareRecordMessagingMessageListenerAdapter<>(
+				this.bean, this.method, this.errorHandler);
+		if (messageConverter instanceof RecordMessageConverter recordMessageConverter) {
+			messageListener.setMessageConverter(recordMessageConverter);
+		}
+		listener = messageListener;
+		if (this.messagingConverter != null) {
+			listener.setMessagingConverter(this.messagingConverter);
+		}
+		BeanResolver resolver = getBeanResolver();
+		if (resolver != null) {
+			listener.setBeanResolver(resolver);
+		}
+		return listener;
+	}
+
 	protected MessagingMessageListenerAdapter<K, V> createMessageListenerInstance(
 			@Nullable MessageConverter messageConverter) {
 

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2025-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.config;
+
+import java.util.Collection;
+
+import org.springframework.kafka.core.ShareConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.ShareKafkaMessageListenerContainer;
+import org.springframework.kafka.support.TopicPartitionOffset;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link KafkaListenerContainerFactory} implementation to create {@link ShareKafkaMessageListenerContainer}
+ * instances for Kafka's share consumer model.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ *
+ * @author Soby Chacko
+ * @since 4.0
+ */
+public class ShareKafkaListenerContainerFactory<K, V>
+		extends AbstractShareKafkaListenerContainerFactory<ShareKafkaMessageListenerContainer<K, V>, K, V> {
+
+	/**
+	 * Construct an instance with the provided consumer factory.
+	 * @param shareConsumerFactory the share consumer factory
+	 */
+	public ShareKafkaListenerContainerFactory(ShareConsumerFactory<K, V> shareConsumerFactory) {
+		setShareConsumerFactory(shareConsumerFactory);
+	}
+
+	@Override
+	protected ShareKafkaMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
+		TopicPartitionOffset[] topicPartitions = endpoint.getTopicPartitionsToAssign();
+		if (topicPartitions != null && topicPartitions.length > 0) {
+			return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(), new ContainerProperties(topicPartitions));
+		}
+		else {
+			Collection<String> topics = endpoint.getTopics();
+			Assert.state(topics != null, "'topics' must not be null");
+			if (!topics.isEmpty()) {
+				return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(),
+						new ContainerProperties(topics.toArray(new String[0])));
+			}
+			else {
+				return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(),
+						new ContainerProperties(endpoint.getTopicPattern()));
+			}
+		}
+	}
+}

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

@@ -21,6 +21,7 @@
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.jspecify.annotations.NonNull;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.BeanUtils;
@@ -58,6 +59,9 @@ public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 	 */
 	public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100;
 
+	/**
+	 * The share consumer factory used to create consumer instances.
+	 */
 	protected final ShareConsumerFactory<K, V> shareConsumerFactory;
 
 	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(this.getClass()));
@@ -66,6 +70,7 @@ public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 
 	protected final ReentrantLock lifecycleLock = new ReentrantLock();
 
+	@NonNull
 	private String beanName = "noBeanNameSet";
 
 	@Nullable

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-present the original author or authors.
+ * Copyright 2025-present the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,8 +43,6 @@
  * This container manages a single-threaded consumer loop using a {@link org.springframework.kafka.core.ShareConsumerFactory}.
  * It is designed for use cases where Kafka's cooperative sharing protocol is desired, and provides a simple polling loop
  * with per-record dispatch and acknowledgement.
- * <p>
- * Lifecycle events are published for consumer starting and started. The container supports direct setting of the client.id.
  *
  * @param <K> the key type
  * @param <V> the value type
@@ -73,7 +71,7 @@ public class ShareKafkaMessageListenerContainer<K, V>
 	 * @param shareConsumerFactory the share consumer factory
 	 * @param containerProperties the container properties
 	 */
-	public ShareKafkaMessageListenerContainer(ShareConsumerFactory<? super K, ? super V> shareConsumerFactory,
+	public ShareKafkaMessageListenerContainer(@Nullable ShareConsumerFactory<? super K, ? super V> shareConsumerFactory,
 			ContainerProperties containerProperties) {
 		super(shareConsumerFactory, containerProperties);
 		Assert.notNull(shareConsumerFactory, "A ShareConsumerFactory must be provided");
@@ -152,7 +150,7 @@ private void publishConsumerStartedEvent() {
 	}
 
 	/**
-	 * The inner share consumer thread: polls for records and dispatches to the listener.
+	 * The inner share consumer thread that polls for records and dispatches to the listener.
 	 */
 	private class ShareListenerConsumer implements Runnable {