Refactor share consumer acknowledgment interfaces and improve code quality

- 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 <soby.chacko@broadcom.com>

Author: sobychacko

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

@@ -231,13 +231,8 @@ private void configureShareGroup(String bootstrapServers, String groupId) throws
 
 Share consumers support two acknowledgment modes that control how records are acknowledged after processing.
 
-[[share-acknowledgment-modes]]
-=== Acknowledgment Modes
-
-Share containers support two acknowledgment modes:
 [[share-implicit-acknowledgment]]
-
-==== Implicit Acknowledgment (Default)
+=== Implicit Acknowledgment (Default)
 In implicit mode, records are automatically acknowledged based on processing outcome:
 
 Successful processing: Records are acknowledged as `ACCEPT`
@@ -259,9 +254,13 @@ public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerCont
 ----
 
 [[share-explicit-acknowledgment]]
-==== Explicit Acknowledgment
+=== Explicit Acknowledgment
+
+In explicit mode, the application must manually acknowledge each record using the provided ShareAcknowledgment.
 
-In explicit mode, the application must manually acknowledge each record using the provided ShareAcknowledgment:
+There are two ways to configure explicit acknowledgment mode:
+
+==== Option 1: Using Kafka Client Configuration
 
 [source,java]
 ----
@@ -271,25 +270,46 @@ public ShareConsumerFactory<String, String> explicitShareConsumerFactory() {
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
-    props.put("share.acknowledgement.mode", "explicit");
+    props.put(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG, "explicit"); // Official Kafka client config
     return new DefaultShareConsumerFactory<>(props);
 }
+----
+
+==== Option 2: Using Spring Container Configuration
 
+[source,java]
+----
 @Bean
 public ShareKafkaListenerContainerFactory<String, String> explicitShareKafkaListenerContainerFactory(
-    ShareConsumerFactory<String, String> explicitShareConsumerFactory) {
-    return new ShareKafkaListenerContainerFactory<>(explicitShareConsumerFactory);
+    ShareConsumerFactory<String, String> shareConsumerFactory) {
+
+    ShareKafkaListenerContainerFactory<String, String> factory =
+        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
+
+    // Configure acknowledgment mode at container factory level
+    factory.getContainerProperties()
+        .setShareAcknowledgmentMode(ContainerProperties.ShareAcknowledgmentMode.EXPLICIT);
+
+    return factory;
 }
 ----
 
+==== Configuration Precedence
+
+When both configuration methods are used, Spring Kafka follows this precedence order (highest to lowest):
+
+1. **Container Properties**: `containerProperties.setShareAcknowledgmentMode()`
+2. **Consumer Config**: `ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG`
+3. **Default**: `ShareAcknowledgmentMode.IMPLICIT`
+
 [[share-acknowledgment-types]]
 === Acknowledgment Types
 
 Share consumers support three acknowledgment types:
 
-`ACCEPT`: Record processed successfully, mark as completed
-`RELEASE`: Temporary failure, make record available for redelivery
-`REJECT`: Permanent failure, do not retry
+ ACCEPT: Record processed successfully, mark as completed
+ RELEASE: Temporary failure, make record available for redelivery
+ REJECT: Permanent failure, do not retry
 
 [[share-acknowledgment-api]]
 === ShareAcknowledgment API
@@ -326,46 +346,60 @@ public void listen(ConsumerRecord<String, String> record) {
 }
 ----
 
-[[share-consumer-aware-listener]]
-==== ShareConsumerAwareMessageListener
+[[share-acknowledging-listener]]
+==== AcknowledgingShareConsumerAwareMessageListener
+
+This interface provides access to the ShareConsumer instance with optional acknowledgment support.
+The acknowledgment parameter is nullable and depends on the container's acknowledgment mode:
 
-Access the ShareConsumer instance for advanced operations:
+===== Implicit Mode Example (acknowledgment is null)
 
 [source,java]
 ----
-@KafkaListener(topics = "my-topic", containerFactory = "shareKafkaListenerContainerFactory")
-public void listen(ConsumerRecord<String, String> record, ShareConsumer<?, ?> consumer) {
+@KafkaListener(
+    topics = "my-topic",
+    containerFactory = "shareKafkaListenerContainerFactory"  // Implicit mode by default
+)
+public void listen(ConsumerRecord<String, String> record,
+                  @Nullable ShareAcknowledgment acknowledgment,
+                  ShareConsumer<?, ?> consumer) {
+
+    // In implicit mode, acknowledgment is null
     System.out.println("Received: " + record.value());
-    // Access consumer metrics, etc.
+
+    // Access consumer metrics if needed
+    Map<MetricName, ? extends Metric> metrics = consumer.metrics();
+
+    // Record is auto-acknowledged as ACCEPT on success, REJECT on error
 }
 ----
 
-[[share-acknowledging-listener]]
-==== AcknowledgingShareConsumerAwareMessageListener
-
-Use explicit acknowledgment with full consumer access:
+===== Explicit Mode Example (acknowledgment is non-null)
 
 [source,java]
 ----
 @Component
 public class ExplicitAckListener {
-@KafkaListener(
-    topics = "my-topic",
-    containerFactory = "explicitShareKafkaListenerContainerFactory"
-)
-public void listen(ConsumerRecord<String, String> record,
-                  ShareAcknowledgment acknowledgment,
-                  ShareConsumer<?, ?> consumer) {
+    @KafkaListener(
+        topics = "my-topic",
+        containerFactory = "explicitShareKafkaListenerContainerFactory"
+    )
+    public void listen(ConsumerRecord<String, String> record,
+                      @Nullable ShareAcknowledgment acknowledgment,
+                      ShareConsumer<?, ?> consumer) {
 
-    try {
-        processRecord(record);
-        acknowledgment.acknowledge(); // ACCEPT
-    } catch (RetryableException e) {
-        acknowledgment.release(); // Will be redelivered
-    } catch (Exception e) {
-        acknowledgment.reject(); // Permanent failure
+        // In explicit mode, acknowledgment is non-null
+        try {
+            processRecord(record);
+            acknowledgment.acknowledge(); // ACCEPT
+        }
+		catch (RetryableException e) {
+            acknowledgment.release(); // Will be redelivered
+        }
+		catch (Exception e) {
+            acknowledgment.reject(); // Permanent failure
+        }
     }
-}
 
     private void processRecord(ConsumerRecord<String, String> record) {
         // Business logic here
@@ -386,6 +420,37 @@ Error Handling: If processing throws an exception, the record is automatically a
 In explicit mode, failing to acknowledge records will block further message processing.
 Always ensure records are acknowledged in all code paths.
 
+[[share-acknowledgment-timeout]]
+==== Acknowledgment Timeout Detection
+
+To help identify missing acknowledgments, Spring Kafka provides configurable timeout detection.
+When a record is not acknowledged within the specified timeout, a warning is logged with details about the unacknowledged record.
+
+[source,java]
+----
+@Bean
+public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerContainerFactory(
+    ShareConsumerFactory<String, String> shareConsumerFactory) {
+    ShareKafkaListenerContainerFactory<String, String> factory =
+        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
+
+    // Set acknowledgment timeout (default is 60 seconds)
+    factory.getContainerProperties().setShareAcknowledgmentTimeout(Duration.ofSeconds(30));
+
+    return factory;
+}
+----
+
+When a record exceeds the timeout, you'll see a warning like:
+----
+WARN: Record not acknowledged within timeout (30 seconds).
+In explicit acknowledgment mode, you must call ack.acknowledge(), ack.release(),
+or ack.reject() for every record.
+Unacknowledged record: topic='my-topic', partition=0, offset=123
+----
+
+This feature helps developers quickly identify when acknowledgment calls are missing from their code, preventing the common issue of "Spring Kafka does not consume new records any more" due to forgotten acknowledgments.
+
 [[share-acknowledgment-examples]]
 === Acknowledgment Examples
 
@@ -406,7 +471,6 @@ Always ensure records are acknowledged in all code paths.
                 else {
                     acknowledgment.release(); // Temporary failure - retry later
                 }
-                }
                 else {
                     acknowledgment.reject(); // Invalid order - don't retry
                 }

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

@@ -98,7 +98,7 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private @Nullable Boolean batchListener;
 
-	private @Nullable  Boolean shareConsumer;
+	private @Nullable Boolean shareConsumer;
 
 	private @Nullable KafkaTemplate<?, ?> replyTemplate;
 
@@ -297,8 +297,13 @@ public void setShareConsumer(Boolean shareConsumer) {
 		this.shareConsumer = shareConsumer;
 	}
 
-	public @Nullable Boolean getShareConsumer() {
-		return this.shareConsumer;
+	/**
+	 * Return true if this endpoint is for a share consumer.
+	 * @return true for a share consumer endpoint.
+	 * @since 4.0
+	 */
+	public boolean isShareConsumer() {
+		return this.shareConsumer != null && this.shareConsumer;
 	}
 
 	/**

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

@@ -211,7 +211,7 @@ protected MessagingMessageListenerAdapter<K, V> createMessageListenerInstance(
 			@Nullable MessageConverter messageConverter) {
 
 		MessagingMessageListenerAdapter<K, V> listener;
-		if (getShareConsumer() != null && getShareConsumer()) {
+		if (isShareConsumer()) {
 			ShareRecordMessagingMessageListenerAdapter<K, V> messageListener = new ShareRecordMessagingMessageListenerAdapter<>(
 					this.bean, this.method, this.errorHandler);
 			if (messageConverter instanceof RecordMessageConverter recordMessageConverter) {

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

@@ -20,6 +20,7 @@
 import java.util.Collection;
 import java.util.regex.Pattern;
 
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.context.ApplicationContext;
@@ -129,38 +130,85 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 		// Validate share group configuration
 		validateShareConfiguration(endpoint);
 
-		Object o = this.shareConsumerFactory.getConfigurationProperties().get("share.acknowledgement.mode");
-		String explicitAck = null;
-		if (o != null) {
-			explicitAck = (String) o;
-		}
+		// Determine acknowledgment mode following Spring Kafka's configuration precedence patterns
+		ContainerProperties.ShareAcknowledgmentMode ackMode = determineAcknowledgmentMode(properties);
+		properties.setShareAcknowledgmentMode(ackMode);
+
 		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)
-				.acceptIfCondition(explicitAck != null && explicitAck.equals("explicit"),
-						ContainerProperties.ShareAcknowledgmentMode.EXPLICIT,
-						properties::setShareAcknowledgmentMode);
+				.acceptIfNotNull(endpoint.getClientIdPrefix(), properties::setClientId);
+	}
+
+	/**
+	 * Determine the acknowledgment mode following Spring Kafka's configuration precedence patterns.
+	 * <p>
+	 * Configuration precedence (highest to lowest):
+	 * <ol>
+	 * <li>Container Properties: {@code containerProperties.getShareAcknowledgmentMode()} (if explicitly set)</li>
+	 * <li>Consumer Config: {@code ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG}</li>
+	 * <li>Default: {@code ShareAcknowledgmentMode.IMPLICIT}</li>
+	 * </ol>
+	 *
+	 * @param containerProperties the container properties to check
+	 * @return the resolved acknowledgment mode
+	 * @throws IllegalArgumentException if an invalid acknowledgment mode is configured
+	 */
+	private ContainerProperties.ShareAcknowledgmentMode determineAcknowledgmentMode(ContainerProperties containerProperties) {
+		// 1. Check if explicitly set at container level (highest priority)
+		// Note: We need to check if it was explicitly set vs using the default
+		// For now, we assume if it's not the default, it was explicitly set
+		ContainerProperties.ShareAcknowledgmentMode containerMode = containerProperties.getShareAcknowledgmentMode();
+		if (containerMode != ContainerProperties.ShareAcknowledgmentMode.IMPLICIT) {
+			// Container level setting takes precedence
+			return containerMode;
+		}
+
+		// 2. Check Kafka client configuration (middle priority)
+		Object clientAckMode = this.shareConsumerFactory.getConfigurationProperties()
+				.get(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
+
+		if (clientAckMode != null) {
+			String mode = clientAckMode.toString();
+			if ("explicit".equals(mode)) {
+				return ContainerProperties.ShareAcknowledgmentMode.EXPLICIT;
+			}
+			else if ("implicit".equals(mode)) {
+				return ContainerProperties.ShareAcknowledgmentMode.IMPLICIT;
+			}
+			else {
+				throw new IllegalArgumentException(
+						"Invalid " + ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG + ": " + mode +
+						". Must be 'implicit' or 'explicit'");
+			}
+		}
+		// 3. Default (lowest priority)
+		return ContainerProperties.ShareAcknowledgmentMode.IMPLICIT;
 	}
 
 	private void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
 		// Validate that batch listeners aren't used with share consumers
-		if (endpoint.getBatchListener() != null && endpoint.getBatchListener()) {
+		if (Boolean.TRUE.equals(endpoint.getBatchListener())) {
 			throw new IllegalArgumentException(
 					"Batch listeners are not supported with share consumers. " +
 							"Share groups operate at the record level.");
 		}
 
-		// Validate acknowledgment mode consistency
+		// Validate acknowledgment mode consistency using official Kafka client configuration
 		Object ackMode = this.shareConsumerFactory.getConfigurationProperties()
-				.get("share.acknowledgement.mode");
-		if (ackMode != null && !Arrays.asList("implicit", "explicit").contains(ackMode)) {
-			throw new IllegalArgumentException(
-					"Invalid share.acknowledgement.mode: " + ackMode +
+				.get(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
+		if (ackMode != null) {
+			String ackModeStr = ackMode.toString().toLowerCase();
+			boolean isValid = Arrays.stream(ContainerProperties.ShareAcknowledgmentMode.values())
+					.anyMatch(mode -> mode.name().toLowerCase().equals(ackModeStr));
+			if (!isValid) {
+				throw new IllegalArgumentException(
+					"Invalid " + ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG + ": " + ackMode +
 							". Must be 'implicit' or 'explicit'");
+			}
 		}
 	}