Simplify share consumer acknowledgment configuration and consolidate 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]>

Author: sobychacko

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

@@ -233,6 +233,7 @@ Share consumers support two acknowledgment modes that control how records are ac
 
 [[share-implicit-acknowledgment]]
 === Implicit Acknowledgment (Default)
+
 In implicit mode, records are automatically acknowledged based on processing outcome:
 
 Successful processing: Records are acknowledged as `ACCEPT`
@@ -245,9 +246,8 @@ public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerCont
     ShareConsumerFactory<String, String> shareConsumerFactory) {
     ShareKafkaListenerContainerFactory<String, String> factory =
     new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
-    // Implicit mode is the default
-    factory.getContainerProperties().setShareAcknowledgmentMode(
-        ContainerProperties.ShareAcknowledgmentMode.IMPLICIT);
+    // Implicit mode is the default (false means implicit acknowledgment)
+    factory.getContainerProperties().setExplicitShareAcknowledgment(false);
 
     return factory;
 }
@@ -287,8 +287,8 @@ public ShareKafkaListenerContainerFactory<String, String> explicitShareKafkaList
         new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
 
     // Configure acknowledgment mode at container factory level
-    factory.getContainerProperties()
-        .setShareAcknowledgmentMode(ContainerProperties.ShareAcknowledgmentMode.EXPLICIT);
+    // true means explicit acknowledgment is required
+    factory.getContainerProperties().setExplicitShareAcknowledgment(true);
 
     return factory;
 }
@@ -298,9 +298,9 @@ public ShareKafkaListenerContainerFactory<String, String> explicitShareKafkaList
 
 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`
+1. **Container Properties**: `containerProperties.setExplicitShareAcknowledgment(true/false)`
+2. **Consumer Config**: `ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG` ("implicit" or "explicit")
+3. **Default**: `false` (implicit acknowledgment)
 
 [[share-acknowledgment-types]]
 === Acknowledgment Types
@@ -319,12 +319,9 @@ The `ShareAcknowledgment` interface provides methods for explicit acknowledgment
 [source,java]
 ----
 public interface ShareAcknowledgment {
-    void acknowledge(AcknowledgeType type);
-    void acknowledge();  // Convenience method for ACCEPT
-    void release();      // Convenience method for RELEASE
-    void reject();       // Convenience method for REJECT
-    boolean isAcknowledged();
-    AcknowledgeType getAcknowledgmentType();
+    void acknowledge();
+    void release();
+    void reject();
 }
 ----
 
@@ -349,7 +346,7 @@ public void listen(ConsumerRecord<String, String> record) {
 [[share-acknowledging-listener]]
 ==== AcknowledgingShareConsumerAwareMessageListener
 
-This interface provides access to the ShareConsumer instance with optional acknowledgment support.
+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:
 
 ===== Implicit Mode Example (acknowledgment is null)
@@ -412,9 +409,9 @@ public class ExplicitAckListener {
 
 In explicit acknowledgment mode, the container enforces important constraints:
 
-Poll Blocking: Subsequent polls are blocked until all records from the previous poll are acknowledged.
-One-time Acknowledgment: Each record can only be acknowledged once.
-Error Handling: If processing throws an exception, the record is automatically acknowledged as `REJECT`.
+ Poll Blocking: Subsequent polls are blocked until all records from the previous poll are acknowledged.
+ One-time Acknowledgment: Each record can only be acknowledged once.
+ Error Handling: If processing throws an exception, the record is automatically acknowledged as `REJECT`.
 
 [WARNING]
 In explicit mode, failing to acknowledge records will block further message processing.
@@ -434,7 +431,7 @@ public ShareKafkaListenerContainerFactory<String, String> shareKafkaListenerCont
     ShareKafkaListenerContainerFactory<String, String> factory =
         new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
 
-    // Set acknowledgment timeout (default is 60 seconds)
+    // Set acknowledgment timeout (default is 30 seconds)
     factory.getContainerProperties().setShareAcknowledgmentTimeout(Duration.ofSeconds(30));
 
     return factory;
@@ -446,7 +443,6 @@ 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.
@@ -504,47 +500,6 @@ public void validateData(ConsumerRecord<String, String> record, ShareAcknowledgm
 }
 ----
 
-[[share-acknowledgment-configuration]]
-=== Acknowledgment Mode Configuration
-
-You can configure the acknowledgment mode at both the consumer factory and container levels:
-
-[[share-factory-level-configuration]]
-==== Factory Level Configuration
-
-[source,java]
-----
-@Bean
-public ShareConsumerFactory<String, String> explicitAckShareConsumerFactory() {
-    Map<String, Object> props = new HashMap<>();
-    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);
-    // Configure explicit acknowledgment at the factory level
-    props.put("share.acknowledgement.mode", "explicit");
-    return new DefaultShareConsumerFactory<>(props);
-}
-----
-
-[[share-container-level-configuration]]
-==== Container Level Configuration
-
-[source,java]
-----
-@Bean
-public ShareKafkaListenerContainerFactory<String, String> customShareKafkaListenerContainerFactory(
-    ShareConsumerFactory<String, String> shareConsumerFactory) {
-    ShareKafkaListenerContainerFactory<String, String> factory =
-        new ShareKafkaListenerContainerFactory<>(shareConsumerFactory);
-
-    // Configure acknowledgment mode at container level
-    factory.getContainerProperties().setShareAcknowledgmentMode(
-    ContainerProperties.ShareAcknowledgmentMode.EXPLICIT);
-
-    return factory;
-}
-----
-
 [[share-differences-from-regular-consumers]]
 == Differences from Regular Consumers
 

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

@@ -293,7 +293,7 @@ public void setBatchListener(boolean batchListener) {
 		this.batchListener = batchListener;
 	}
 
-	public void setShareConsumer(Boolean shareConsumer) {
+	public void setShareConsumer(boolean shareConsumer) {
 		this.shareConsumer = shareConsumer;
 	}
 

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

@@ -131,8 +131,8 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 		validateShareConfiguration(endpoint);
 
 		// Determine acknowledgment mode following Spring Kafka's configuration precedence patterns
-		ContainerProperties.ShareAcknowledgmentMode ackMode = determineAcknowledgmentMode(properties);
-		properties.setShareAcknowledgmentMode(ackMode);
+		boolean explicitAck = determineExplicitAcknowledgment(properties);
+		properties.setExplicitShareAcknowledgment(explicitAck);
 
 		JavaUtils.INSTANCE
 				.acceptIfNotNull(effectiveAutoStartup, instance::setAutoStartup)
@@ -144,49 +144,39 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 	}
 
 	/**
-	 * Determine the acknowledgment mode following Spring Kafka's configuration precedence patterns.
+	 * Determine whether explicit acknowledgment is required 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>Container Properties: {@code containerProperties.isExplicitShareAcknowledgment()} (if explicitly set)</li>
 	 * <li>Consumer Config: {@code ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG}</li>
-	 * <li>Default: {@code ShareAcknowledgmentMode.IMPLICIT}</li>
+	 * <li>Default: {@code false} (implicit acknowledgment)</li>
 	 * </ol>
-	 *
 	 * @param containerProperties the container properties to check
-	 * @return the resolved acknowledgment mode
+	 * @return true if explicit acknowledgment is required, false for implicit
 	 * @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)
+	private boolean determineExplicitAcknowledgment(ContainerProperties containerProperties) {
+		// Check Kafka client configuration
 		Object clientAckMode = this.shareConsumerFactory.getConfigurationProperties()
 				.get(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
 
 		if (clientAckMode != null) {
-			String mode = clientAckMode.toString();
+			String mode = clientAckMode.toString().toLowerCase();
 			if ("explicit".equals(mode)) {
-				return ContainerProperties.ShareAcknowledgmentMode.EXPLICIT;
+				return true;
 			}
 			else if ("implicit".equals(mode)) {
-				return ContainerProperties.ShareAcknowledgmentMode.IMPLICIT;
+				return false;
 			}
 			else {
 				throw new IllegalArgumentException(
 						"Invalid " + ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG + ": " + mode +
 						". Must be 'implicit' or 'explicit'");
 			}
 		}
-		// 3. Default (lowest priority)
-		return ContainerProperties.ShareAcknowledgmentMode.IMPLICIT;
+		// Default to implicit acknowledgment (false)
+		return containerProperties.isExplicitShareAcknowledgment();
 	}
 
 	private void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
@@ -197,19 +187,6 @@ private void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
 							"Share groups operate at the record level.");
 		}
 
-		// Validate acknowledgment mode consistency using official Kafka client configuration
-		Object ackMode = this.shareConsumerFactory.getConfigurationProperties()
-				.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'");
-			}
-		}
 	}
 
 	@Override

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

@@ -47,7 +47,6 @@
  *
  * @see ShareAcknowledgment
  * @see ShareConsumer
- * @see ContainerProperties.ShareAcknowledgmentMode
  */
 @FunctionalInterface
 public interface AcknowledgingShareConsumerAwareMessageListener<K, V> extends GenericMessageListener<ConsumerRecord<K, V>> {

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

@@ -161,21 +161,6 @@ public enum EOSMode {
 
 	}
 
-	/**
-	 * Acknowledgment mode for share consumer containers.
-	 */
-	public enum ShareAcknowledgmentMode {
-		/**
-		 * Records are automatically acknowledged as ACCEPT on next poll, commitSync, or commitAsync.
-		 */
-		IMPLICIT,
-
-		/**
-		 * Application must explicitly acknowledge all records before next poll.
-		 */
-		EXPLICIT
-	}
-
 	/**
 	 * The default {@link #setShutdownTimeout(long) shutDownTimeout} (ms).
 	 */
@@ -329,9 +314,9 @@ public enum ShareAcknowledgmentMode {
 
 	private boolean recordObservationsInBatch;
 
-	private ShareAcknowledgmentMode shareAcknowledgmentMode = ShareAcknowledgmentMode.IMPLICIT;
+	private boolean explicitShareAcknowledgment = false;
 
-	private Duration shareAcknowledgmentTimeout = Duration.ofSeconds(60); // 1 minute default
+	private Duration shareAcknowledgmentTimeout = Duration.ofSeconds(30); // Align with Kafka's share.record.lock.duration.ms default
 
 	/**
 	 * Create properties for a container that will subscribe to the specified topics.
@@ -1136,34 +1121,30 @@ public void setRecordObservationsInBatch(boolean recordObservationsInBatch) {
 	}
 
 	/**
-	 * Set the acknowledgment mode for share consumer containers.
+	 * Set whether explicit acknowledgment is required for share consumer containers.
 	 * <p>
 	 * This setting only applies to share consumer containers and is ignored
-	 * by regular consumer containers. The acknowledgment mode determines
-	 * how records are acknowledged:
-	 * <ul>
-	 * <li>{@link ShareAcknowledgmentMode#IMPLICIT} - Records are automatically
-	 * acknowledged as ACCEPT when the next poll occurs or when commitSync/commitAsync
-	 * is called</li>
-	 * <li>{@link ShareAcknowledgmentMode#EXPLICIT} - Application must explicitly
-	 * acknowledge each record using the provided {@link ShareAcknowledgment}</li>
-	 * </ul>
-	 *
-	 * @param shareAcknowledgmentMode the acknowledgment mode
+	 * by regular consumer containers.
+	 * <p>
+	 * When set to {@code false} (default), records are automatically acknowledged
+	 * as ACCEPT when the next poll occurs or when commitSync/commitAsync is called.
+	 * <p>
+	 * When set to {@code true}, the application must explicitly acknowledge each
+	 * record using the provided {@link ShareAcknowledgment}.
+	 * @param explicitShareAcknowledgment true for explicit acknowledgment, false for implicit
 	 * @since 4.0
 	 * @see ShareAcknowledgment
 	 */
-	public void setShareAcknowledgmentMode(ShareAcknowledgmentMode shareAcknowledgmentMode) {
-		this.shareAcknowledgmentMode = shareAcknowledgmentMode;
+	public void setExplicitShareAcknowledgment(boolean explicitShareAcknowledgment) {
+		this.explicitShareAcknowledgment = explicitShareAcknowledgment;
 	}
 
 	/**
-	 * Get the acknowledgment mode for share consumer containers.
-	 *
-	 * @return the acknowledgment mode
+	 * Check whether explicit acknowledgment is required for share consumer containers.
+	 * @return true if explicit acknowledgment is required, false for implicit acknowledgment
 	 */
-	public ShareAcknowledgmentMode getShareAcknowledgmentMode() {
-		return this.shareAcknowledgmentMode;
+	public boolean isExplicitShareAcknowledgment() {
+		return this.explicitShareAcknowledgment;
 	}
 
 	/**
@@ -1173,8 +1154,7 @@ public ShareAcknowledgmentMode getShareAcknowledgmentMode() {
 	 * will be logged to help identify missing acknowledgment calls.
 	 * This only applies when using explicit acknowledgment mode.
 	 * <p>
-	 * Default is 60 seconds.
-	 *
+	 * Default is 30 seconds.
 	 * @param shareAcknowledgmentTimeout the timeout duration
 	 * @since 4.0
 	 */
@@ -1184,7 +1164,6 @@ public void setShareAcknowledgmentTimeout(Duration shareAcknowledgmentTimeout) {
 
 	/**
 	 * Get the timeout for share acknowledgments in explicit mode.
-	 *
 	 * @return the acknowledgment timeout
 	 * @since 4.0
 	 */

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

@@ -69,11 +69,11 @@
  *
  * @param <K> the key type
  * @param <V> the value type
+ *
  * @author Soby Chacko
  * @since 4.0
  * @see ShareConsumer
  * @see ShareAcknowledgment
- * @see ContainerProperties.ShareAcknowledgmentMode
  */
 public class ShareKafkaMessageListenerContainer<K, V>
 		extends AbstractShareKafkaMessageListenerContainer<K, V> {
@@ -148,6 +148,16 @@ protected void doStart() {
 		}
 		GenericMessageListener<?> listener = (GenericMessageListener<?>) messageListener;
 		Assert.state(listener != null, "'messageListener' cannot be null");
+
+		// Validate listener type for explicit acknowledgment mode
+		if (containerProperties.isExplicitShareAcknowledgment()) {
+			boolean isAcknowledgingListener = listener instanceof AcknowledgingShareConsumerAwareMessageListener;
+			Assert.state(isAcknowledgingListener,
+					"Explicit acknowledgment mode requires an AcknowledgingShareConsumerAwareMessageListener. " +
+					"Current listener type: " + listener.getClass().getName() + ". " +
+					"Either use implicit acknowledgment mode or provide a listener that can handle acknowledgments.");
+		}
+
 		this.listenerConsumer = new ShareListenerConsumer(listener);
 		setRunning(true);
 		this.listenerConsumerFuture = CompletableFuture.runAsync(this.listenerConsumer, consumerExecutor);
@@ -230,8 +240,7 @@ private class ShareListenerConsumer implements Runnable {
 			ContainerProperties containerProperties = getContainerProperties();
 
 			// Configure acknowledgment mode
-			this.isExplicitMode = containerProperties.getShareAcknowledgmentMode() ==
-					ContainerProperties.ShareAcknowledgmentMode.EXPLICIT;
+			this.isExplicitMode = containerProperties.isExplicitShareAcknowledgment();
 			this.ackTimeoutMs = containerProperties.getShareAcknowledgmentTimeout().toMillis();
 
 			// Configure consumer properties based on acknowledgment mode