Add Kafka shared consumer container support #3945

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Merged

artembilan merged 3 commits into spring-projects:main from sobychacko:shared-consumer-iteration-3

Jun 6, 2025

Contributor

sobychacko commented Jun 5, 2025

New AbstractShareKafkaMessageListenerContainer base class with lifecycle management
ShareKafkaMessageListenerContainer implementation for share consumer protocol
Integration tests for end-to-end message delivery validation

sobychacko added this to the 4.0.0-M3 milestone

sobychacko added the type: enhancement label

artembilan requested changes

View reviewed changes

Member

artembilan left a comment

I don't have too many concerns.
Just regular review.
Thanks

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

+              	 */
+              	public static final int DEFAULT_PHASE = Integer.MAX_VALUE - 100;
+              	@NonNull

Member

artembilan Jun 5, 2025

I think you are missing a package-info.java file over here with that our @org.jspecify.annotations.NullMarked to avoid this annotation explicitly.

...main/java/org/springframework/kafka/listener/AbstractShareKafkaMessageListenerContainer.java Show resolved Hide resolved

...fka/src/main/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainer.java Show resolved Hide resolved

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

+              		ShareListenerConsumer(GenericMessageListener<?> listener) {
+              			this.consumer = ShareKafkaMessageListenerContainer.this.shareConsumerFactory.createShareConsumer(
+              				ShareKafkaMessageListenerContainer.this.getGroupId(),

Member

artembilan Jun 5, 2025

Just nit-pick: I prefer to have a blank line after method subject if it is multi-line.
Otherwise it is hard to read the method body.
Well, just my humble opinion.

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

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

Member

artembilan Jun 5, 2025

Why fully-qualified class name?
Why not import for this java.util.Arrays?

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

+              			Throwable exitThrowable = null;
+              			while (isRunning()) {
+              				try {
+              					var records = this.consumer.poll(java.time.Duration.ofMillis(1000));

Member

artembilan Jun 5, 2025

I guess the poll duration could be a property of this class.

...fka/src/main/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainer.java Show resolved Hide resolved

...a/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerIntegrationTests.java

+              				new ConfigResource(ConfigResource.Type.GROUP, groupId), List.of(op));
+              		try (Admin admin = AdminClient.create(adminProperties)) {
+              			admin.incrementalAlterConfigs(configs).all().get();
+              		}

Member

artembilan Jun 5, 2025

This is a bit awkward approach to just set that share.auto.offset.reset property.
Cannot it be done via our brokerProperties on the @EmbeddedKafka?
How this suppose to happen in the target applications?
It feels like changing the broker state from the application is not OK.
Especially when we have several instances looking into the same Kafka cluster.

Contributor Author

sobychacko Jun 5, 2025

I think in the current release of Kafka (4.0.0), this is how it is done, i.e. we need to set it at the groupId. I don't think we can set it via broker properties in EmbeddedKafka. Maybe that will change once this feature is stabilized and production ready in 4.1. See this for details: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=255070434#KIP932:QueuesforKafka-Groupconfiguration

Member

artembilan commented Jun 5, 2025

 > Could not resolve all files for configuration ':spring-kafka:compileClasspath'.
   > Could not find io.projectreactor.kafka:reactor-kafka:.
     Required by:
         project :spring-kafka

I guess time to deprecate those Reactor Kafka components in 3.3.x and remove altogether here in 4.0.
Looks like reactor-kafka is not a part of reactor-bom any more.

Contributor Author

sobychacko commented Jun 5, 2025

Ok, i will send a PR soon.

artembilan requested changes

View reviewed changes

Member

artembilan left a comment

> Could not resolve all files for configuration ':spring-kafka:compileClasspath'.
   > Could not find io.projectreactor.kafka:reactor-kafka:.
     Required by:
         project :spring-kafka

I guess this PR has to be rebased to the latest main.

Thanks

sobychacko added 3 commits

June 6, 2025 11:13


          Add Kafka shared consumer container support

ec59f25

- New AbstractShareKafkaMessageListenerContainer base class with lifecycle management
- ShareKafkaMessageListenerContainer implementation for share consumer protocol
- Integration tests for end-to-end message delivery validation

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


          Addressing PR review

f06ef14


          checkstyle

4c1d5e1

sobychacko force-pushed the shared-consumer-iteration-3 branch from 619d89d to 4c1d5e1 Compare

June 6, 2025 15:15

artembilan merged commit 5aa4175 into spring-projects:main

2 of 3 checks passed

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

type: enhancement