Add comprehensive acknowledgment support for Kafka share consumers #4087

sobychacko · 2025-09-15T01:45:13Z

Implement explicit and implicit acknowledgment modes for share consumer containers, enabling fine-grained control over record processing outcomes with ACCEPT, RELEASE, and REJECT acknowledgment types.

Add ShareAcknowledgment interface with ACCEPT/RELEASE/REJECT support
Add ShareAcknowledgmentException for acknowledgment failures
Implement ShareAcknowledgmentMode enum (IMPLICIT/EXPLICIT) in ContainerProperties
Add poll-level acknowledgment constraints in explicit mode
Add ShareConsumerAwareMessageListener for ShareConsumer access
Add AcknowledgingShareConsumerAwareMessageListener for explicit acknowledgment
Add ShareRecordMessagingMessageListenerAdapter for KafkaListener integration
Use non-polymorphic onShareRecord method names to avoid regression issues with existing listener infrastructure and maintain clear API separation
Enhanced ShareKafkaMessageListenerContainer with acknowledgment tracking
Automatic error handling with REJECT acknowledgment on exceptions
Poll blocking in explicit mode until all records acknowledged
Support for mixed acknowledgment patterns within single poll
Auto-detection of ShareKafkaListenerContainerFactory for share consumer endpoints
Validation preventing batch listeners with share consumers
Factory-level and container-level acknowledgment mode configuration
Message converter extensions for ShareAcknowledgment parameter injection
Comprehensive integration tests covering all acknowledgment scenarios
Constraint tests validating poll-level acknowledgment requirements
Unit tests for container behavior and listener dispatching
Updated documentation with acknowledgment examples
Implicit: Records auto-acknowledged as ACCEPT on success, REJECT on error
Explicit: Application must manually acknowledge each record before next poll
Explicit mode blocks subsequent polls until all records acknowledged
Prevents message loss and ensures proper acknowledgment ordering
Concurrent acknowledgment attempts properly handled with IllegalStateException
Processing exceptions trigger automatic REJECT acknowledgment
Acknowledgment failures reset state and throw ShareAcknowledgmentException
Container continues processing after individual record failures
ShareAcknowledgment parameter injection follows existing Spring Kafka patterns
Non-polymorphic listener method names (onShareRecord vs onMessage) prevent potential conflicts with existing listener infrastructure and ensure clear separation between regular and share consumer listener contracts
Factory and container level configuration options provide flexibility

This implementation provides acknowledgment semantics for Kafka share groups while maintaining backward compatibility with existing implicit acknowledgment behavior.

artembilan

The review so far.
Might be the case that some of it is void according to your new vision, but still might be helpful.

Thanks

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

.../java/org/springframework/kafka/listener/AcknowledgingShareConsumerAwareMessageListener.java

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

...afka/src/main/java/org/springframework/kafka/listener/ShareConsumerAwareMessageListener.java

artembilan

Sorry, the review is too long.
But thanks for doing this not easy stuff!

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

...rc/main/java/org/springframework/kafka/listener/adapter/MessagingMessageListenerAdapter.java

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

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessageConverter.java

...va/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerConstraintTests.java

...est/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerUnitTests.java

artembilan

Another review round.

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

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

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

artembilan · 2025-09-29T19:40:09Z

...va/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerConstraintTests.java

+			produceTestRecords(bootstrapServers, topic, 2);
+
+			// Wait and verify second batch is NOT processed yet
+			Thread.sleep(3000);


I don't think this is reliable way to delay.
Please, consider to use some other async barrier instead.

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

artembilan · 2025-09-29T19:50:25Z

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

+			assertThat(pendingAcks).hasSize(3);
+
+			// Wait a bit to ensure no more records are processed while acknowledgments are pending
+			Thread.sleep(2000);


Cannot we ensure ordering instead of vague timeout?

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

...va/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerConstraintTests.java

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

Implement explicit and implicit acknowledgment modes for share consumer containers, enabling fine-grained control over record processing outcomes with ACCEPT, RELEASE, and REJECT acknowledgment types. - Add ShareAcknowledgment interface with ACCEPT/RELEASE/REJECT support - Add ShareAcknowledgmentException for acknowledgment failures - Implement ShareAcknowledgmentMode enum (IMPLICIT/EXPLICIT) in ContainerProperties - Add poll-level acknowledgment constraints in explicit mode - Add ShareConsumerAwareMessageListener for ShareConsumer access - Add AcknowledgingShareConsumerAwareMessageListener for explicit acknowledgment - Add ShareRecordMessagingMessageListenerAdapter for KafkaListener integration - Use non-polymorphic onShareRecord method names to avoid regression issues with existing listener infrastructure and maintain clear API separation - Enhanced ShareKafkaMessageListenerContainer with acknowledgment tracking - Automatic error handling with REJECT acknowledgment on exceptions - Poll blocking in explicit mode until all records acknowledged - Support for mixed acknowledgment patterns within single poll - Auto-detection of ShareKafkaListenerContainerFactory for share consumer endpoints - Validation preventing batch listeners with share consumers - Factory-level and container-level acknowledgment mode configuration - Message converter extensions for ShareAcknowledgment parameter injection - Comprehensive integration tests covering all acknowledgment scenarios - Constraint tests validating poll-level acknowledgment requirements - Unit tests for container behavior and listener dispatching - Updated documentation with acknowledgment examples - Implicit: Records auto-acknowledged as ACCEPT on success, REJECT on error - Explicit: Application must manually acknowledge each record before next poll - Explicit mode blocks subsequent polls until all records acknowledged - Prevents message loss and ensures proper acknowledgment ordering - Concurrent acknowledgment attempts properly handled with IllegalStateException - Processing exceptions trigger automatic REJECT acknowledgment - Acknowledgment failures reset state and throw ShareAcknowledgmentException - Container continues processing after individual record failures - ShareAcknowledgment parameter injection follows existing Spring Kafka patterns - Non-polymorphic listener method names (onShareRecord vs onMessage) prevent potential conflicts with existing listener infrastructure and ensure clear separation between regular and share consumer listener contracts - Factory and container level configuration options provide flexibility This implementation provides acknowledgment semantics for Kafka share groups while maintaining backward compatibility with existing implicit acknowledgment behavior. Signed-off-by: Soby Chacko <[email protected]>

…ality - Simplify to single AcknowledgingShareConsumerAwareMessageListener interface with nullable acknowledgment - Remove redundant ShareConsumerAwareMessageListener interface - Clean up ShareAcknowledgmentMode enum by removing unnecessary string property - Replace verbose null checks with Boolean.TRUE.equals() pattern - Use enum values for acknowledgment mode validation instead of hardcoded strings - Update tests and documentation to use unified interface - Fix spacing issues in @nullable annotations - Fixing other formatting issues Improve share consumer polling behavior and add isShareConsumer() helper - Add isShareConsumer() helper method to AbstractKafkaListenerEndpoint for cleaner boolean checks - Replace verbose Boolean.TRUE.equals() pattern with cleaner isShareConsumer() call - Handle KIP-932 IllegalStateException when polling with unacknowledged records in explicit mode - Add minimal 10ms delay to prevent tight loop while maintaining responsiveness - Remove problematic pre-poll blocking logic that prevented proper exception handling The share consumer now properly handles the broker's IllegalStateException when attempting to poll with unacknowledged records, as specified in KIP-932. This maintains heartbeat while waiting for acknowledgments and prevents CPU-intensive tight loops. Fix thread safety in ShareConsumer acknowledgments ShareConsumer is not thread-safe and requires all access to happen on the consumer thread. The previous implementation allowed acknowledgment calls from listener threads to directly access the consumer, causing ConcurrentModificationException. Changes: - Add PendingAcknowledgment queue to safely pass acknowledgments between threads - Process queued acknowledgments on the consumer thread during poll loop - Remove direct consumer access from ShareConsumerAcknowledgment.acknowledgeInternal() - Add notifyAcknowledged() callback for acknowledgment completion This ensures all ShareConsumer interactions happen on the owning consumer thread, eliminating race conditions between polling and acknowledgment operations. The thread safety issue was exposed by removing the pre-poll sleep, which previously masked the concurrent access by creating timing windows where the consumer was dormant. Signed-off-by: Soby Chacko <[email protected]>

…message converters - Replace ShareAcknowledgmentMode enum with boolean explicitShareAcknowledgment field - Consolidate duplicate toMessage/toShareMessage methods into single toMessage with Object parameters - Merge invoke/invokeHandler methods in MessagingMessageListenerAdapter to use Object parameters - Eliminate duplicate commonHeaders methods in MessageConverter interface - Add validation for explicit acknowledgment mode requiring proper listener interface - Simplify ShareKafkaMessageListenerContainerUnitTests to focus on configuration validation - Other formatting fixes in docs and javadocs Signed-off-by: Soby Chacko <[email protected]>

- Replace Thread.sleep() with CountDownLatch assertions in tests for reliable synchronization and faster test execution - Use LogMessage.format() instead of Supplier with String.format() for consistent logging patterns - Fix lambda expression to use effectively final variable in error logging - Change shareConsumer field from nullable Boolean to primitive boolean in AbstractKafkaListenerEndpoint for cleaner null handling - Make utility methods static where appropriate for better code organization Signed-off-by: Soby Chacko <[email protected]>

constraint and integration tests, as testing for non-occurrence (await expecting false) still blocks for full timeout duration - Use chained assertion style for acknowledgment validations Signed-off-by: Soby Chacko <[email protected]>

- Replace Thread.sleep() with Awaitility pattern that verifies poll blocking via trace log - Add Mockito spy on LogAccessor to confirm the logging message - Migrate shouldHandlePartialAcknowledgmentCorrectly test from ConstraintTests - Migrate shouldHandleConcurrentAcknowledgmentAttempts test from ConstraintTests - Fix topic collision by assigning unique topics to each test - Remove duplicate ShareKafkaMessageListenerContainerConstraintTests class This improves test reliability by verifying actual blocking behavior rather than assuming Thread.sleep duration is sufficient, and eliminates test duplication. Signed-off-by: Soby Chacko <[email protected]>

artembilan

That's it.
Everything else is just great!

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

artembilan · 2025-10-02T15:13:02Z

...main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

 				containerFactory, beanName);

+		if (listenerContainerFactory instanceof ShareKafkaListenerContainerFactory<?, ?>) {
+			endpoint.setShareConsumer(Boolean.TRUE);


Please, address.

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

artembilan · 2025-10-02T15:19:57Z

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

+
+		if (clientAckMode != null) {
+			String mode = clientAckMode.toString().toLowerCase();
+			if ("explicit".equals(mode)) {


Can we rely on the ShareAcknowledgementMode.fromString() API instead of the manual logic in this method?

artembilan · 2025-10-02T15:33:16Z

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

+import org.springframework.kafka.listener.KafkaListenerErrorHandler;
+import org.springframework.kafka.support.ShareAcknowledgment;
+import org.springframework.kafka.support.converter.JacksonProjectingMessageConverter;
+import org.springframework.kafka.support.converter.ProjectingMessageConverter;


I prefer to avoid deprecated imports and use FQCN instead of the place of use where you already have that @SuppressWarnings("removal")

artembilan · 2025-10-02T15:36:45Z

spring-kafka/src/main/java/org/springframework/kafka/support/ShareAcknowledgmentException.java

+ * @author Soby Chacko
+ * @since 4.0
+ */
+public class ShareAcknowledgmentException extends KafkaException {


Looks like this class is out of use.
Perhaps just remove?

@SuppressWarnings

- Change autoStartup and phase from nullable wrappers to primitives with defaults - Add @SuppressWarnings(NullAway.Init) to applicationEventPublisher and applicationContext - Use ShareAcknowledgementMode.fromString() API instead of manual string parsing - Use ArgumentMatchers with explicit type parameters to eliminate unchecked warnings in tests - Remove deprecated imports from ShareRecordMessagingMessageListenerAdapter - Use FQCN for JacksonProjectingMessageConverter and ProjectingMessageConverter - Fix syntax error in kafka-queues.adoc documentation example Signed-off-by: Soby Chacko <[email protected]>

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

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

 			RecordMessageConverter messageConverter = getMessageConverter();
-			if (!(messageConverter instanceof JacksonProjectingMessageConverter
-					|| messageConverter instanceof ProjectingMessageConverter)) {
+			if (!(messageConverter instanceof org.springframework.kafka.support.converter.JacksonProjectingMessageConverter


This is an overreacting.
I have talked only about a deprecated ProjectingMessageConverter class.

Related to: spring-projects/spring-kafka#4087 In particular the `MessageConverter` in SK has a new signature from now on: ``` toMessage(ConsumerRecord<?, ?> record, Object acknowledgment, Object consumer, Type type); ```

artembilan added this to the 4.0.0-RC1 milestone Sep 18, 2025

artembilan added the type: enhancement label Sep 18, 2025

artembilan requested changes Sep 18, 2025

View reviewed changes

sobychacko force-pushed the share-group-acknowledgment-changes branch from 178aedf to 11bf13e Compare September 24, 2025 20:11

artembilan requested changes Sep 26, 2025

View reviewed changes

artembilan requested changes Sep 29, 2025

View reviewed changes

artembilan requested changes Sep 30, 2025

View reviewed changes

...va/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerConstraintTests.java Outdated Show resolved Hide resolved

...a/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerIntegrationTests.java Outdated Show resolved Hide resolved

sobychacko added 5 commits October 1, 2025 15:53

sobychacko force-pushed the share-group-acknowledgment-changes branch from 37a23ac to 6bc2806 Compare October 2, 2025 14:41

sobychacko force-pushed the share-group-acknowledgment-changes branch from 6bc2806 to 43b8c43 Compare October 2, 2025 14:42

artembilan requested changes Oct 2, 2025

View reviewed changes

artembilan requested changes Oct 3, 2025

View reviewed changes

cleanup from PR review

9c8e7a4

artembilan merged commit 7e64584 into spring-projects:main Oct 3, 2025
2 of 3 checks passed

Add comprehensive acknowledgment support for Kafka share consumers #4087

Add comprehensive acknowledgment support for Kafka share consumers #4087

Uh oh!

Conversation

sobychacko commented Sep 15, 2025

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

artembilan Sep 29, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan Sep 29, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan Oct 2, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan Oct 2, 2025

Choose a reason for hiding this comment

Uh oh!

artembilan Oct 2, 2025

Choose a reason for hiding this comment

Uh oh!

artembilan Oct 2, 2025

Choose a reason for hiding this comment

Uh oh!

artembilan Oct 3, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!