Add KafkaListener support for shared consumer containers

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

@@ -0,0 +1,171 @@
+/*
+ * 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.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.kafka.core.ShareConsumerFactory;
+import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.ShareKafkaMessageListenerContainer;
+import org.springframework.kafka.support.JavaUtils;
+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.
+ * <p>
+ * This 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 <K> the key type
+ * @param <V> the value type
+ *
+ * @author Soby Chacko
+ * @since 4.0
+ */
+public class ShareKafkaListenerContainerFactory<K, V>
+		implements KafkaListenerContainerFactory<ShareKafkaMessageListenerContainer<K, V>>, ApplicationEventPublisherAware, ApplicationContextAware {
+
+	private final ShareConsumerFactory<? super K, ? super V> shareConsumerFactory;
+
+	private @Nullable Boolean autoStartup;
+
+	private @Nullable Integer phase;
+
+	private @Nullable ApplicationEventPublisher applicationEventPublisher;
+
+	private @Nullable ApplicationContext applicationContext;
+
+	/**
+	 * Construct an instance with the provided consumer factory.
+	 * @param shareConsumerFactory the share consumer factory
+	 */
+	public ShareKafkaListenerContainerFactory(ShareConsumerFactory<K, V> shareConsumerFactory) {
+		this.shareConsumerFactory = shareConsumerFactory;
+	}
+
+	@Override
+	public void setApplicationContext(ApplicationContext applicationContext) {
+		this.applicationContext = applicationContext;
+	}
+
+	/**
+	 * 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;
+	}
+
+	@Override
+	@SuppressWarnings({"unchecked", "rawtypes"})
+	public ShareKafkaMessageListenerContainer<K, V> createListenerContainer(KafkaListenerEndpoint endpoint) {
+		ShareKafkaMessageListenerContainer<K, V> instance = createContainerInstance(endpoint);
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
+		if (endpoint instanceof AbstractKafkaListenerEndpoint abstractKafkaListenerEndpoint) {
+			configureEndpoint(abstractKafkaListenerEndpoint);
+		}
+		// TODO: No message converter for queue at the moment
+		endpoint.setupListenerContainer(instance, null);
+		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(ShareKafkaMessageListenerContainer<K, V> instance, KafkaListenerEndpoint endpoint) {
+		ContainerProperties properties = instance.getContainerProperties();
+		Boolean effectiveAutoStartup = endpoint.getAutoStartup() != null ? endpoint.getAutoStartup() : this.autoStartup;
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(effectiveAutoStartup, instance::setAutoStartup)
+				.acceptIfNotNull(this.phase, instance::setPhase)
+				.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
+				.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
+				.acceptIfNotNull(endpoint.getGroupId(), properties::setGroupId)
+				.acceptIfNotNull(endpoint.getClientIdPrefix(), properties::setClientId)
+				.acceptIfNotNull(endpoint.getConsumerProperties(), properties::setKafkaConsumerProperties);
+	}
+
+	@Override
+	public ShareKafkaMessageListenerContainer<K, V> createContainer(TopicPartitionOffset... topicPartitions) {
+		throw new UnsupportedOperationException("ShareConsumer does not support explicit partition assignment");
+	}
+
+	@Override
+	public ShareKafkaMessageListenerContainer<K, V> createContainer(String... topics) {
+		return createContainerInstance(new KafkaListenerEndpointAdapter() {
+
+			@Override
+			public Collection<String> getTopics() {
+				return Arrays.asList(topics);
+			}
+		});
+	}
+
+	@Override
+	public ShareKafkaMessageListenerContainer<K, V> createContainer(Pattern topicPattern) {
+		throw new UnsupportedOperationException("ShareConsumer does not support topic patterns");
+	}
+
+	/**
+	 * Create a container instance for the provided endpoint.
+	 * @param endpoint the endpoint
+	 * @return the container instance
+	 */
+	protected ShareKafkaMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
+		Collection<String> topics = endpoint.getTopics();
+		Assert.state(topics != null, "'topics' must not be null");
+		return new ShareKafkaMessageListenerContainer<>(this.shareConsumerFactory,
+				new ContainerProperties(topics.toArray(new String[0])));
+	}
+
+}

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

@@ -58,6 +58,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()));
@@ -86,7 +89,7 @@ public abstract class AbstractShareKafkaMessageListenerContainer<K, V>
 	 * @param containerProperties the properties.
 	 */
 	@SuppressWarnings("unchecked")
-	protected AbstractShareKafkaMessageListenerContainer(@Nullable ShareConsumerFactory<? super K, ? super V> shareConsumerFactory,
+	protected AbstractShareKafkaMessageListenerContainer(ShareConsumerFactory<? super K, ? super V> shareConsumerFactory,
 														ContainerProperties containerProperties) {
 		Assert.notNull(containerProperties, "'containerProperties' cannot be null");
 		Assert.notNull(shareConsumerFactory, "'shareConsumerFactory' cannot be null");

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.
@@ -23,6 +23,7 @@
 import java.util.concurrent.CountDownLatch;
 
 import org.apache.kafka.clients.consumer.AcknowledgeType;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ShareConsumer;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
@@ -43,8 +44,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
@@ -79,14 +78,6 @@ public ShareKafkaMessageListenerContainer(ShareConsumerFactory<? super K, ? supe
 		Assert.notNull(shareConsumerFactory, "A ShareConsumerFactory must be provided");
 	}
 
-	/**
-	 * Set the {@code client.id} to use for the consumer.
-	 * @param clientId the client id to set
-	 */
-	public void setClientId(String clientId) {
-		this.clientId = clientId;
-	}
-
 	/**
 	 * Get the {@code client.id} for the consumer.
 	 * @return the client id, or null if not set
@@ -96,6 +87,14 @@ public String getClientId() {
 		return this.clientId;
 	}
 
+	/**
+	 * Set the {@code client.id} to use for the consumer.
+	 * @param clientId the client id to set
+	 */
+	public void setClientId(String clientId) {
+		this.clientId = clientId;
+	}
+
 	@Override
 	public boolean isInExpectedState() {
 		return isRunning();
@@ -152,7 +151,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 {
 
@@ -168,12 +167,11 @@ private class ShareListenerConsumer implements Runnable {
 
 		ShareListenerConsumer(GenericMessageListener<?> listener) {
 			this.consumer = ShareKafkaMessageListenerContainer.this.shareConsumerFactory.createShareConsumer(
-				ShareKafkaMessageListenerContainer.this.getGroupId(),
-				ShareKafkaMessageListenerContainer.this.getClientId());
+					ShareKafkaMessageListenerContainer.this.getGroupId(),
+					ShareKafkaMessageListenerContainer.this.getClientId());
 
 			this.genericListener = listener;
 			this.clientId = ShareKafkaMessageListenerContainer.this.getClientId();
-			// Subscribe to topics, just like in the test
 			ContainerProperties containerProperties = getContainerProperties();
 			this.consumer.subscribe(Arrays.asList(containerProperties.getTopics()));
 		}
@@ -184,6 +182,7 @@ String getClientId() {
 		}
 
 		@Override
+		@SuppressWarnings({"unchecked", "rawtypes"})
 		public void run() {
 			initialize();
 			Throwable exitThrowable = null;
@@ -192,9 +191,14 @@ public void run() {
 					var records = this.consumer.poll(java.time.Duration.ofMillis(POLL_TIMEOUT));
 					if (records != null && records.count() > 0) {
 						for (var record : records) {
-							@SuppressWarnings("unchecked")
-							GenericMessageListener<Object> listener = (GenericMessageListener<Object>) this.genericListener;
-							listener.onMessage(record);
+							if (this.genericListener instanceof AcknowledgingConsumerAwareMessageListener ackListener) {
+								ackListener.onMessage(record, null, null);
+							}
+							else {
+								GenericMessageListener<ConsumerRecord<K, V>> listener =
+										(GenericMessageListener<ConsumerRecord<K, V>>) this.genericListener;
+								listener.onMessage(record);
+							}
 							// Temporarily auto-acknowledge and commit.
 							// We will refactor it later on to support more production-like scenarios.
 							this.consumer.acknowledge(record, AcknowledgeType.ACCEPT);
@@ -235,9 +239,11 @@ private void wrapUp() {
 		@Override
 		public String toString() {
 			return "ShareKafkaMessageListenerContainer.ShareListenerConsumer ["
-				+ "consumerGroupId=" + this.consumerGroupId
-				+ ", clientId=" + this.clientId
-				+ "]";
+					+ "consumerGroupId=" + this.consumerGroupId
+					+ ", clientId=" + this.clientId
+					+ "]";
 		}
+
 	}
+
 }