Add KafkaListener support for shared consumer containers #3988

sobychacko · 2025-07-01T02:27:23Z

Add AbstractShareKafkaListenerContainerFactory base class for share consumer factories
Add ShareKafkaListenerContainerFactory concrete implementation
Add ShareRecordMessagingMessageListenerAdapter for share consumer message handling
Modify MethodKafkaListenerEndpoint to create appropriate listener adapters based on container type
Add integration tests for ShareKafkaListener functionality

- Add AbstractShareKafkaListenerContainerFactory base class for share consumer factories - Add ShareKafkaListenerContainerFactory concrete implementation - Add ShareRecordMessagingMessageListenerAdapter for share consumer message handling - Modify MethodKafkaListenerEndpoint to create appropriate listener adapters based on container type - Add integration tests for ShareKafkaListener functionality Signed-off-by: Soby Chacko <[email protected]>

artembilan

I don't know too many details about the feature, so forgive me if my review has gone sideways.

artembilan · 2025-07-01T17:44:34Z

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

+ * @since 4.0
+ */
+public class ShareKafkaListenerContainerFactory<K, V>
+		extends AbstractShareKafkaListenerContainerFactory<ShareKafkaMessageListenerContainer<K, V>, K, V> {


Any idea why do we need an abstract super class instead of combining all the logic in this class?

artembilan · 2025-07-01T17:48:50Z

...main/java/org/springframework/kafka/listener/AbstractShareKafkaMessageListenerContainer.java


 	protected final ReentrantLock lifecycleLock = new ReentrantLock();

+	@NonNull


No one ever use this annotation if we do @org.jspecify.annotations.NullMarked in the package-info.java.
That one assumes that all not marked with @Nullable properties are not null.
If you have a complain here that it might be null, then we need to revise the logic around this property.
However I believe it is good enough to mark it as @SuppressWarnings("NullAway.Init") because this property is usually initialized from the application context.

artembilan · 2025-07-01T17:49:59Z

...fka/src/main/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainer.java

 	 * @param containerProperties the container properties
 	 */
-	public ShareKafkaMessageListenerContainer(ShareConsumerFactory<? super K, ? super V> shareConsumerFactory,
+	public ShareKafkaMessageListenerContainer(@Nullable ShareConsumerFactory<? super K, ? super V> shareConsumerFactory,


Why have this happened?
How did that turned out that now a ShareKafkaMessageListenerContainer can be created without a factory?

artembilan · 2025-07-01T17:52:58Z

...a/org/springframework/kafka/listener/adapter/ShareRecordMessagingMessageListenerAdapter.java

+	 */
+	@Override
+	public void onMessage(ConsumerRecord<K, V> record) {
+		onMessage(record, null, null);


I don't see a reason in this class at all.
What is wrong with using RecordMessagingMessageListenerAdapter as it is?

...fka/src/test/java/org/springframework/kafka/listener/ShareKafkaListenerIntegrationTests.java

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

artembilan · 2025-07-02T15:17:10Z

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

-		MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);
+
+		final MessagingMessageListenerAdapter<K, V> messageListener;
+		messageListener = createMessageListenerInstance(messageConverter);


Let's revise changes in this class!
What you have done on these lines makes no sense so far.
The previous code version was much better to read, and it is identical to the logic you have so far.

ya, that was an oversight. will fix it.

artembilan · 2025-07-02T15:17:37Z

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

-	 * Create an empty {@link MessagingMessageListenerAdapter} instance.
-	 * @param messageConverter the converter (may be null).
-	 * @return the {@link MessagingMessageListenerAdapter} instance.
-	 */


Not sure if this is OK to remove this JavaDoc.
More explanations - better.

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

artembilan · 2025-07-02T15:28:35Z

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

+	protected ShareKafkaMessageListenerContainer<K, V> createContainerInstance(KafkaListenerEndpoint endpoint) {
+		TopicPartitionOffset[] topicPartitions = endpoint.getTopicPartitionsToAssign();
+		if (topicPartitions != null && topicPartitions.length > 0) {
+			return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(), new ContainerProperties(topicPartitions));


And probably this one is not covered by tests.
So far we have there in the ShareKafkaMessageListenerContainer only this:

// Subscribe to topics, just like in the test ContainerProperties containerProperties = getContainerProperties(); this.consumer.subscribe(Arrays.asList(containerProperties.getTopics()));

Not sure why you say just like in the test though. Probably just remove that comment?

artembilan · 2025-07-02T15:30:26Z

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

+			}
+			else {
+				return new ShareKafkaMessageListenerContainer<>(getShareConsumerFactory(),
+						new ContainerProperties(endpoint.getTopicPattern()));


Probably KafkaShareConsumer does not support pattern subscription as well...
Please, revise.

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

artembilan · 2025-07-02T19:57:18Z

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

 				"Could not create message listener - MessageHandlerMethodFactory not set");
-		MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);
+
+		final MessagingMessageListenerAdapter<K, V> messageListener = createMessageListenerInstance(messageConverter);


Still some unnecessary change for me to review? 😉

artembilan · 2025-07-02T19:58:03Z

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

+		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
+			configureEndpoint((AbstractKafkaListenerEndpoint<K, V>) endpoint);
+		}
+		endpoint.setupListenerContainer(instance, null); // No message converter for MVP


What is MVP, please?

artembilan · 2025-07-02T20:05:34Z

...fka/src/test/java/org/springframework/kafka/listener/ShareKafkaListenerIntegrationTests.java

+			received.set(record.value());
+			latch.countDown();
+		}
+	}


Consider to use src/idea/spring-framework.xml for your IntelliJ IDEA editor config.
Then reformat classes, like this and ShareKafkaMessageListenerContainer.
So, all those missed blank lines around class members are going to be added.
Therefore, next time when I do some changes in the class would be consistent with what you've done to it.

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

artembilan · 2025-07-03T14:32:46Z

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

+public class ShareKafkaListenerContainerFactory<K, V>
+		implements KafkaListenerContainerFactory<ShareKafkaMessageListenerContainer<K, V>>, ApplicationEventPublisherAware, ApplicationContextAware {
+
+	protected final LogAccessor logger = new LogAccessor(LogFactory.getLog(getClass()));


Doesn't look like it has to be protected.
Also nit-pick, there is this ctor:

/** * Create a new accessor for the specified Commons Log category. * @see LogFactory#getLog(Class) */ public LogAccessor(Class<?> logCategory) {

artembilan · 2025-07-03T14:34:32Z

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

+	 * Get the container properties.
+	 * @return the container properties
+	 */
+	public ContainerProperties getContainerProperties() {


Why do we need getter for this?

I even wonder if we need this.containerProperties at all...

artembilan · 2025-07-03T14:35:10Z

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

+		ShareKafkaMessageListenerContainer<K, V> instance = createContainerInstance(endpoint);
+		JavaUtils.INSTANCE
+				.acceptIfNotNull(endpoint.getId(), instance::setBeanName);
+		if (endpoint instanceof AbstractKafkaListenerEndpoint) {


This could be pattern expression:

if (endpoint instanceof AbstractKafkaListenerEndpoint abstractKafkaListenerEndpoint) {

And not cast in the next line.

artembilan · 2025-07-03T14:35:48Z

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

+	 * Get the share consumer factory.
+	 * @return the share consumer factory
+	 */
+	public ShareConsumerFactory<? super K, ? super V> getShareConsumerFactory() {


Why do we need getter for this?

artembilan · 2025-07-03T14:36:16Z

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

+	 * 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) {


Why do we need setter for this if we request it from the ctor?

artembilan · 2025-07-03T14:37:08Z

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

+		}
+		if (endpoint.getConsumerProperties() != null) {
+			instance.getContainerProperties().setKafkaConsumerProperties(endpoint.getConsumerProperties());
+		}


Let's see if the logic of this method could be reworked to JavaUtils API.

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

artembilan · 2025-07-03T17:02:06Z

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

+					if (endpoint.getAutoStartup() == null) {
+						instance.setAutoStartup(autoStartup);
+					}
+				})


Some is off with these two calls.
Both of them are about calling instance.setAutoStartup() in the end.
Can we calculate a single autoStartup variable before propagating it down?

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

sobychacko force-pushed the KL-shared-consumer branch from 3bcce66 to 69f4e3b Compare July 1, 2025 02:29

artembilan requested changes Jul 1, 2025

View reviewed changes

Addressing PR review

8ae7ff9

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

artembilan requested changes Jul 2, 2025

View reviewed changes

Addressing PR review

20978de

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

artembilan requested changes Jul 2, 2025

View reviewed changes

Addressing PR review

e9ee474

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

artembilan requested changes Jul 3, 2025

View reviewed changes

Addressing PR review

d0f6965

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

artembilan requested changes Jul 3, 2025

View reviewed changes

Addressing PR review

8829ac9

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

artembilan added this to the 4.0.0-M3 milestone Jul 3, 2025

artembilan added the type: enhancement label Jul 3, 2025

artembilan enabled auto-merge (squash) July 3, 2025 17:11

artembilan merged commit de1622d into spring-projects:main Jul 3, 2025
3 checks passed


		protected final ReentrantLock lifecycleLock = new ReentrantLock();

		@NonNull

Add KafkaListener support for shared consumer containers #3988

Add KafkaListener support for shared consumer containers #3988

Uh oh!

Conversation

sobychacko commented Jul 1, 2025

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!