Introduce ShareAckMode enum for share consumers

The 4.0 boolean `explicitShareAcknowledgment` only controlled whether
the listener received a non-null acknowledgment object. It had no
effect on Kafka's `share.acknowledgement.mode`, so `acknowledge()`
would throw `IllegalStateException` when the factory used the Kafka
default (implicit mode).

`ContainerProperties.ShareAckMode` (EXPLICIT, MANUAL, IMPLICIT) makes
the modes explicit, with EXPLICIT as the default — the same philosophy
as regular containers disabling auto.commit and owning the commit
lifecycle themselves. For EXPLICIT and MANUAL the container now
enforces `share.acknowledgement.mode=explicit` via consumer override
properties, regardless of factory configuration.

Other related fixes:

 - `RecordDeserializationException` handler now guards `acknowledge()`
   against IMPLICIT mode
 - `createShareConsumer` variant added to `ShareConsumerFactory` for
   consumer-level overrides
 - 4.0 dead code and naming issues cleaned up

Deprecated methods keep the 4.0 API intact.
Reference docs and What's New include a migration guide.
Adding tests to verify the changes.

Signed-off-by: Soby Chacko <soby.chacko@broadcom.com>

Author: sobychacko

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

@@ -14,6 +14,63 @@ The `@KafkaListener` annotation now supports an `ackMode` attribute, allowing in
 The attribute also supports SpEL expressions and property placeholders.
 See xref:kafka/receiving-messages/listener-annotation.adoc[`@KafkaListener` Annotation] for more information.
 
+[[x41-share-ack-mode]]
+=== Share Consumer Acknowledgment Modes
+
+The boolean `setExplicitShareAcknowledgment(boolean)` property on `ContainerProperties` has been replaced by the `ShareAckMode` enum, which clearly names the three distinct acknowledgment modes:
+
+* `EXPLICIT` (default) — Container-managed.
+The container sends `ACCEPT` after successful processing and delegates error handling to the `ShareConsumerRecordRecoverer` (default: `REJECT`).
+This is the analogue of disabling `auto.commit` on a regular consumer.
+* `MANUAL` — Listener-managed.
+The listener must acknowledge each record via the provided `ShareAcknowledgment`.
+Subsequent polls are blocked until all records from the previous poll are acknowledged.
+* `IMPLICIT` — Kafka client implicit mode.
+The broker auto-accepts all records regardless of processing outcome.
+
+Configure the mode on the factory:
+
+[source,java]
+----
+factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);
+----
+
+The deprecated `setExplicitShareAcknowledgment(true)` maps to `MANUAL`; `setExplicitShareAcknowledgment(false)` maps to `EXPLICIT`.
+See xref:kafka/kafka-queues.adoc#share-record-acknowledgment[Record Acknowledgment] for the full reference.
+
+==== Migration Guide
+
+**Default behavior is unchanged.**
+The old `setExplicitShareAcknowledgment(false)` default was already container-managed acknowledgment (the container sent `ACCEPT` on success), which is exactly what `ShareAckMode.EXPLICIT` does.
+No action is required for applications using the default.
+
+**If you used `setExplicitShareAcknowledgment(true)`**, replace it:
+
+[source,java]
+----
+// Before (4.0)
+factory.getContainerProperties().setExplicitShareAcknowledgment(true);
+
+// After (4.1)
+factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.MANUAL);
+----
+
+**If you set `share.acknowledgement.mode=implicit` in the factory configuration** (via `ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG`), this is a breaking change.
+In 4.0 this setting had no effect because the container always called `consumer.acknowledge()` regardless, which would have thrown `IllegalStateException` in true Kafka implicit mode.
+In 4.1, the container detects this conflict and logs a warning, then overrides the factory setting with explicit mode.
+To genuinely use Kafka client implicit mode — where the broker auto-accepts all records regardless of processing outcome — you must now opt in explicitly:
+
+[source,java]
+----
+factory.getContainerProperties().setShareAckMode(ContainerProperties.ShareAckMode.IMPLICIT);
+----
+
+[WARNING]
+====
+In `ShareAckMode.IMPLICIT`, the `ShareConsumerRecordRecoverer` is not consulted and processing errors do not influence acknowledgment.
+Records are always ACCEPTed by the broker.
+====
+
 [[x41-share-consumer-error-handling]]
 === Share Consumer Error Handling
 

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

@@ -20,8 +20,6 @@
 import java.util.Collection;
 import java.util.regex.Pattern;
 
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.internals.ShareAcknowledgementMode;
 import org.jspecify.annotations.Nullable;
 
 import org.springframework.beans.BeanUtils;
@@ -183,11 +181,6 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 		BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",
 				"messageListener", "ackCount", "ackTime", "subBatchPerPartition", "kafkaConsumerProperties");
 
-		// Determine acknowledgment mode following Spring Kafka's configuration precedence patterns
-		// Check factory-level properties first, then consumer factory config
-		boolean explicitAck = determineExplicitAcknowledgment(properties);
-		properties.setExplicitShareAcknowledgment(explicitAck);
-
 		// Set concurrency - endpoint setting takes precedence over factory setting
 		Integer conc = endpoint.getConcurrency();
 		if (conc != null) {
@@ -208,39 +201,6 @@ protected void initializeContainer(ShareKafkaMessageListenerContainer<K, V> inst
 				.acceptIfNotNull(endpoint.getClientIdPrefix(), properties::setClientId);
 	}
 
-	/**
-	 * 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.isExplicitShareAcknowledgment()} (if explicitly set via factory-level properties)</li>
-	 * <li>Consumer Config: {@code ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG}</li>
-	 * <li>Default: {@code false} (implicit acknowledgment)</li>
-	 * </ol>
-	 * @param containerProperties the container properties to check
-	 * @return true if explicit acknowledgment is required, false for implicit
-	 * @throws IllegalArgumentException if an invalid acknowledgment mode is configured
-	 */
-	private boolean determineExplicitAcknowledgment(ContainerProperties containerProperties) {
-		// Check factory-level properties first
-		// If explicitly set to true (non-default), use it with highest precedence
-		if (this.containerProperties.isExplicitShareAcknowledgment()) {
-			return true;
-		}
-
-		// Check Kafka client configuration as fallback
-		Object clientAckMode = this.shareConsumerFactory.getConfigurationProperties()
-				.get(ConsumerConfig.SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
-
-		if (clientAckMode != null) {
-			ShareAcknowledgementMode mode = ShareAcknowledgementMode.fromString(clientAckMode.toString());
-			return mode == ShareAcknowledgementMode.EXPLICIT;
-		}
-
-		// Default to implicit acknowledgment (false)
-		return false;
-	}
-
 	private static void validateShareConfiguration(KafkaListenerEndpoint endpoint) {
 		// Validate that batch listeners aren't used with share consumers
 		if (Boolean.TRUE.equals(endpoint.getBatchListener())) {

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

@@ -26,6 +26,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaShareConsumer;
 import org.apache.kafka.clients.consumer.ShareConsumer;
 import org.apache.kafka.common.MetricName;
@@ -141,19 +142,42 @@ public ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullab
 		return createRawConsumer(groupId, clientId);
 	}
 
+	@Override
+	public ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId,
+			Map<String, Object> overrideProperties) {
+		return createRawConsumer(groupId, clientId, overrideProperties);
+	}
+
 	/**
-	 * Actually create the consumer.
+	 * Create the consumer with no override properties.
+	 * Subclasses may override to customize consumer creation.
 	 * @param groupId the group id (maybe null).
-	 * @param clientId the client id.
+	 * @param clientId the client id (maybe null).
 	 * @return the share consumer.
 	 */
 	protected ShareConsumer<K, V> createRawConsumer(@Nullable String groupId, @Nullable String clientId) {
+		return createRawConsumer(groupId, clientId, Collections.emptyMap());
+	}
+
+	/**
+	 * Actually create the consumer, applying override properties on top of the
+	 * factory configuration. Override properties take precedence over factory
+	 * configuration; {@code groupId} and {@code clientId} are applied last.
+	 * @param groupId the group id (maybe null).
+	 * @param clientId the client id (maybe null).
+	 * @param overrideProperties properties to apply on top of the factory configuration.
+	 * @return the share consumer.
+	 * @since 4.1
+	 */
+	protected ShareConsumer<K, V> createRawConsumer(@Nullable String groupId, @Nullable String clientId,
+			Map<String, Object> overrideProperties) {
 		Map<String, Object> consumerProperties = new HashMap<>(this.configs);
+		consumerProperties.putAll(overrideProperties);
 		if (groupId != null) {
-			consumerProperties.put("group.id", groupId);
+			consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
 		}
 		if (clientId != null) {
-			consumerProperties.put("client.id", clientId);
+			consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
 		}
 		return new ExtendedShareConsumer(consumerProperties);
 	}

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultShareConsumerFactory.java

@@ -43,6 +43,25 @@ public interface ShareConsumerFactory<K, V> {
 	 */
 	ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId);
 
+	/**
+	 * Create a share consumer with the provided group id, client id, and
+	 * additional properties that override the factory's configuration.
+	 * The container uses this to enforce internal configuration (e.g. acknowledgement
+	 * mode) without mutating the factory's configuration.
+	 * Implementations that do not override this method will fall back to
+	 * {@link #createShareConsumer(String, String)}, and the override properties
+	 * will be ignored.
+	 * @param groupId the group id (maybe null).
+	 * @param clientId the client id (maybe null).
+	 * @param overrideProperties properties to apply on top of the factory configuration.
+	 * @return the share consumer.
+	 * @since 4.1
+	 */
+	default ShareConsumer<K, V> createShareConsumer(@Nullable String groupId, @Nullable String clientId,
+			Map<String, Object> overrideProperties) {
+		return createShareConsumer(groupId, clientId);
+	}
+
 	/**
 	 * Return an unmodifiable reference to the configuration map for this factory.
 	 * Useful for cloning to make a similar factory.

spring-kafka/src/main/java/org/springframework/kafka/core/ShareConsumerFactory.java

@@ -27,16 +27,20 @@
  * <p>
  * This interface provides access to both the {@link ShareConsumer} instance and acknowledgment
  * capabilities. The acknowledgment parameter behavior depends on the container's
- * acknowledgment mode:
+ * {@link ContainerProperties.ShareAckMode}:
  * <ul>
- * <li><strong>Explicit mode</strong>: The acknowledgment parameter is non-null and must
- * be used to acknowledge each record</li>
- * <li><strong>Implicit mode</strong>: The acknowledgment parameter is null and records
- * are automatically acknowledged</li>
+ * <li><strong>MANUAL</strong>: The acknowledgment is non-null; the listener must call
+ * {@link org.springframework.kafka.support.ShareAcknowledgment#acknowledge()},
+ * {@link org.springframework.kafka.support.ShareAcknowledgment#release()}, or
+ * {@link org.springframework.kafka.support.ShareAcknowledgment#reject()} for every record</li>
+ * <li><strong>EXPLICIT</strong>: The acknowledgment is null; the container sends ACCEPT
+ * automatically on success and delegates errors to the {@link ShareConsumerRecordRecoverer}</li>
+ * <li><strong>IMPLICIT</strong>: The acknowledgment is null; the broker auto-accepts all records</li>
  * </ul>
  * <p>
  * This is the primary listener interface for share consumers when you need access
- * to the ShareConsumer instance or need explicit acknowledgment control.
+ * to the {@link ShareConsumer} instance or need listener-managed acknowledgment
+ * control ({@code ShareAckMode.MANUAL}).
  *
  * @param <K> the key type
  * @param <V> the value type
@@ -52,12 +56,11 @@
 public interface AcknowledgingShareConsumerAwareMessageListener<K, V> extends GenericMessageListener<ConsumerRecord<K, V>> {
 
 	/**
-	 * Invoked with data from kafka, an acknowledgment, and provides access to the consumer.
-	 * When explicit acknowledgment mode is used, the acknowledgment parameter will be non-null
-	 * and must be used to acknowledge the record. When implicit acknowledgment mode is used,
-	 * the acknowledgment parameter will be null.
+	 * Invoked with data from Kafka, an acknowledgment, and provides access to the consumer.
+	 * The acknowledgment is non-null only in {@link ContainerProperties.ShareAckMode#MANUAL} mode;
+	 * it is null in {@code EXPLICIT} and {@code IMPLICIT} modes.
 	 * @param data the data to be processed.
-	 * @param acknowledgment the acknowledgment (nullable in implicit mode).
+	 * @param acknowledgment the acknowledgment, or {@code null} if not in MANUAL mode.
 	 * @param consumer the consumer.
 	 */
 	void onShareRecord(ConsumerRecord<K, V> data, @Nullable ShareAcknowledgment acknowledgment, ShareConsumer<?, ?> consumer);

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

@@ -35,7 +35,6 @@
 import org.springframework.aop.framework.ProxyFactory;
 import org.springframework.aop.support.AopUtils;
 import org.springframework.core.task.AsyncTaskExecutor;
-import org.springframework.kafka.support.ShareAcknowledgment;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.micrometer.KafkaListenerObservationConvention;
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
@@ -117,6 +116,40 @@ public enum AckMode {
 
 	}
 
+	/**
+	 * The acknowledgment mode for share consumer containers.
+	 * @since 4.1
+	 * @see #setShareAckMode(ShareAckMode)
+	 */
+	public enum ShareAckMode {
+
+		/**
+		 * Kafka client implicit mode. All records are automatically acknowledged
+		 * as ACCEPT by the broker regardless of processing outcome. No per-record
+		 * acknowledgment control is available in this mode. Equivalent to setting
+		 * {@code share.acknowledgement.mode=implicit} on the Kafka client.
+		 */
+		IMPLICIT,
+
+		/**
+		 * Kafka client explicit mode, container-managed. The container sends ACCEPT
+		 * after successful processing and delegates to the
+		 * {@link ShareConsumerRecordRecoverer} (default: REJECT) on error.
+		 * This is the default.
+		 */
+		EXPLICIT,
+
+		/**
+		 * Kafka client explicit mode, listener-managed. The listener must acknowledge
+		 * each record manually via the provided
+		 * {@link org.springframework.kafka.support.ShareAcknowledgment}.
+		 * Subsequent polls are blocked until all records from the previous poll
+		 * are acknowledged.
+		 */
+		MANUAL
+
+	}
+
 	/**
 	 * Offset commit behavior during assignment.
 	 * @since 2.3.6
@@ -314,7 +347,7 @@ public enum EOSMode {
 
 	private boolean recordObservationsInBatch;
 
-	private boolean explicitShareAcknowledgment = false;
+	private ShareAckMode shareAckMode = ShareAckMode.EXPLICIT; // default: container-managed explicit mode
 
 	private Duration shareAcknowledgmentTimeout = Duration.ofSeconds(30); // Align with Kafka's share.record.lock.duration.ms default
 
@@ -1121,38 +1154,54 @@ public void setRecordObservationsInBatch(boolean recordObservationsInBatch) {
 	}
 
 	/**
-	 * Set whether explicit acknowledgment is required for share consumer containers.
+	 * Set the acknowledgment mode for share consumer containers.
 	 * <p>
 	 * This setting only applies to share consumer containers and is ignored
 	 * 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
+	 * @param shareAckMode the acknowledgment mode; default {@link ShareAckMode#EXPLICIT}.
+	 * @since 4.1
+	 * @see ShareAckMode
+	 */
+	public void setShareAckMode(ShareAckMode shareAckMode) {
+		this.shareAckMode = shareAckMode;
+	}
+
+	/**
+	 * Return the acknowledgment mode for share consumer containers.
+	 * @return the acknowledgment mode.
+	 * @since 4.1
+	 */
+	public ShareAckMode getShareAckMode() {
+		return this.shareAckMode;
+	}
+
+	/**
+	 * Set whether to use explicit share acknowledgment mode.
+	 * @param explicitShareAcknowledgment {@code true} to use {@link ShareAckMode#MANUAL},
+	 * {@code false} to use {@link ShareAckMode#EXPLICIT}.
+	 * @deprecated in favor of {@link #setShareAckMode(ShareAckMode)} with
+	 * {@link ShareAckMode#MANUAL}.
 	 */
+	@Deprecated(since = "4.1", forRemoval = false)
 	public void setExplicitShareAcknowledgment(boolean explicitShareAcknowledgment) {
-		this.explicitShareAcknowledgment = explicitShareAcknowledgment;
+		this.shareAckMode = explicitShareAcknowledgment ? ShareAckMode.MANUAL : ShareAckMode.EXPLICIT;
 	}
 
 	/**
-	 * Check whether explicit acknowledgment is required for share consumer containers.
-	 * @return true if explicit acknowledgment is required, false for implicit acknowledgment
+	 * Return whether the current mode is {@link ShareAckMode#MANUAL}.
+	 * @return {@code true} if the current {@link ShareAckMode} is {@link ShareAckMode#MANUAL}.
+	 * @deprecated in favor of {@link #getShareAckMode()}.
 	 */
+	@Deprecated(since = "4.1", forRemoval = false)
 	public boolean isExplicitShareAcknowledgment() {
-		return this.explicitShareAcknowledgment;
+		return this.shareAckMode == ShareAckMode.MANUAL;
 	}
 
 	/**
-	 * Set the timeout for share acknowledgments in explicit mode.
+	 * Set the timeout for share acknowledgments in {@link ShareAckMode#MANUAL} mode.
 	 * <p>
-	 * When a record is not acknowledged within this timeout, a warning
-	 * will be logged to help identify missing acknowledgment calls.
-	 * This only applies when using explicit acknowledgment mode.
+	 * When a record is not acknowledged within this timeout, a warning will be logged
+	 * to help identify missing acknowledgment calls. Only applies to {@code MANUAL} mode.
 	 * <p>
 	 * Default is 30 seconds.
 	 * @param shareAcknowledgmentTimeout the timeout duration
@@ -1163,7 +1212,7 @@ public void setShareAcknowledgmentTimeout(Duration shareAcknowledgmentTimeout) {
 	}
 
 	/**
-	 * Get the timeout for share acknowledgments in explicit mode.
+	 * Get the timeout for share acknowledgments in {@link ShareAckMode#MANUAL} mode.
 	 * @return the acknowledgment timeout
 	 * @since 4.0
 	 */