Deprecate autoCommitOffset in favor of ackMode (#957)

* Deprecate autoCommitOffset in favor or ackMode

Deprecate autoCommitOffset in favor of using a newly introduced consumer property ackMode.
If the consumer is not in batch mode and if ackEachRecord is enabled, then container
will use RECORD ackMode. Otherwise, use the provided ackMode using this property.
If none of these are true, then it will defer to the default setting of BATCH ackMode
set by the container.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/877

* Address PR review comments

* Addressing PR review comments

docs/src/main/asciidoc/overview.adoc

@@ -180,6 +180,8 @@ By default, offsets are committed after all records in the batch of records retu
 The number of records returned by a poll can be controlled with the `max.poll.records` Kafka property, which is set through the consumer `configuration` property.
 Setting this to `true` may cause a degradation in performance, but doing so reduces the likelihood of redelivered records when a failure occurs.
 Also, see the binder `requiredAcks` property, which also affects the performance of committing offsets.
+This property is deprecated as of 3.1 in favor of using `ackMode`.
+If the `ackMode` is not set and batch mode is not enabled, `RECORD` ackMode will be used.
 +
 Default: `false`.
 autoCommitOffset::
@@ -188,9 +190,14 @@ If set to `false`, a header with the key `kafka_acknowledgment` of the type `org
 Applications may use this header for acknowledging messages.
 See the examples section for details.
 When this property is set to `false`, Kafka binder sets the ack mode to `org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode.MANUAL` and the application is responsible for acknowledging records.
-Also see `ackEachRecord`.
+Also see `ackEachRecord`. This property is deprecated as of 3.1. See `ackMode` for more details.
 +
 Default: `true`.
+ackMode::
+Specify the container ack mode.
+This is based on the AckMode enumeration defined in Spring Kafka.
+If `ackEachRecord` property is set to `true` and consumer is not in batch mode, then this will use the ack mode of `RECORD`, otherwise, use the provided ack mode using this property.
+
 autoCommitOnError::
 Effective only if `autoCommitOffset` is set to `true`.
 If set to `false`, it suppresses auto-commits for messages that result in errors and commits only for successful messages. It allows a stream to automatically replay from the last successfully processed message, in case of persistent failures.

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java

@@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.properties;
 import java.util.HashMap;
 import java.util.Map;
 
+import org.springframework.kafka.listener.ContainerProperties;
+
 /**
  * Extended consumer properties for Kafka binder.
  *
@@ -88,6 +90,7 @@ public class KafkaConsumerProperties {
 	 * When true the offset is committed after each record, otherwise the offsets for the complete set of records
 	 * received from the poll() are committed after all records have been processed.
 	 */
+	@Deprecated
 	private boolean ackEachRecord;
 
 	/**
@@ -101,8 +104,15 @@ public class KafkaConsumerProperties {
 	 * If set to false, a header with the key kafka_acknowledgment of the type org.springframework.kafka.support.Acknowledgment header
 	 * is present in the inbound message. Applications may use this header for acknowledging messages.
 	 */
+	@Deprecated
 	private boolean autoCommitOffset = true;
 
+	/**
+	 * Controlling the container acknowledgement mode. This is the preferred way to control the ack mode on the
+	 * container instead of the deprecated autoCommitOffset property.
+	 */
+	private ContainerProperties.AckMode ackMode;
+
 	/**
 	 * Effective only if autoCommitOffset is set to true.
 	 * If set to false, it suppresses auto-commits for messages that result in errors and commits only for successful messages.
@@ -111,6 +121,7 @@ public class KafkaConsumerProperties {
 	 * If not set (the default), it effectively has the same value as enableDlq,
 	 * auto-committing erroneous messages if they are sent to a DLQ and not committing them otherwise.
 	 */
+	@Deprecated
 	private Boolean autoCommitOnError;
 
 	/**
@@ -205,11 +216,20 @@ public class KafkaConsumerProperties {
 	 *
 	 * When true the offset is committed after each record, otherwise the offsets for the complete set of records
 	 * received from the poll() are committed after all records have been processed.
+	 *
+	 * @deprecated since 3.1 in favor of using {@link #ackMode}
 	 */
+	@Deprecated
 	public boolean isAckEachRecord() {
 		return this.ackEachRecord;
 	}
 
+	/**
+	 * @param ackEachRecord
+	 *
+	 * @deprecated in favor of using {@link #ackMode}
+	 */
+	@Deprecated
 	public void setAckEachRecord(boolean ackEachRecord) {
 		this.ackEachRecord = ackEachRecord;
 	}
@@ -220,15 +240,35 @@ public class KafkaConsumerProperties {
 	 * Whether to autocommit offsets when a message has been processed.
 	 * If set to false, a header with the key kafka_acknowledgment of the type org.springframework.kafka.support.Acknowledgment header
 	 * is present in the inbound message. Applications may use this header for acknowledging messages.
+	 *
+	 * @deprecated since 3.1 in favor of using {@link #ackMode}
 	 */
+	@Deprecated
 	public boolean isAutoCommitOffset() {
 		return this.autoCommitOffset;
 	}
 
+	/**
+	 * @param autoCommitOffset
+	 *
+	 * @deprecated in favor of using {@link #ackMode}
+	 */
+	@Deprecated
 	public void setAutoCommitOffset(boolean autoCommitOffset) {
 		this.autoCommitOffset = autoCommitOffset;
 	}
 
+	/**
+	 * @return Container's ack mode.
+	 */
+	public ContainerProperties.AckMode getAckMode() {
+		return this.ackMode;
+	}
+
+	public void setAckMode(ContainerProperties.AckMode ackMode) {
+		this.ackMode = ackMode;
+	}
+
 	/**
 	 * @return start offset
 	 *
@@ -280,11 +320,21 @@ public class KafkaConsumerProperties {
 	 * If set to true, it always auto-commits (if auto-commit is enabled).
 	 * If not set (the default), it effectively has the same value as enableDlq,
 	 * auto-committing erroneous messages if they are sent to a DLQ and not committing them otherwise.
+	 *
+	 * @deprecated in favor of using an error handler and customize the container with that error handler.
 	 */
+	@Deprecated
 	public Boolean getAutoCommitOnError() {
 		return this.autoCommitOnError;
 	}
 
+	/**
+	 *
+	 * @param autoCommitOnError commit on error
+	 *
+	 * @deprecated in favor of using an error handler and customize the container with that error handler.
+	 */
+	@Deprecated
 	public void setAutoCommitOnError(Boolean autoCommitOnError) {
 		this.autoCommitOnError = autoCommitOnError;
 	}

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -655,19 +655,18 @@ public class KafkaMessageChannelBinder extends
 		}
 		messageListenerContainer.setBeanName(destination + ".container");
 		// end of these won't be needed...
-		if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) {
-			messageListenerContainer.getContainerProperties()
-					.setAckMode(ContainerProperties.AckMode.MANUAL);
-			messageListenerContainer.getContainerProperties().setAckOnError(false);
+		ContainerProperties.AckMode ackMode = extendedConsumerProperties.getExtension().getAckMode();
+		if (ackMode == null && extendedConsumerProperties.getExtension().isAckEachRecord()) {
+			ackMode = ContainerProperties.AckMode.RECORD;
 		}
-		else {
-			messageListenerContainer.getContainerProperties()
-					.setAckOnError(isAutoCommitOnError(extendedConsumerProperties));
-			if (extendedConsumerProperties.getExtension().isAckEachRecord()) {
+		if (ackMode != null) {
+			if ((extendedConsumerProperties.isBatchMode() && ackMode != ContainerProperties.AckMode.RECORD) ||
+					!extendedConsumerProperties.isBatchMode()) {
 				messageListenerContainer.getContainerProperties()
-						.setAckMode(ContainerProperties.AckMode.RECORD);
+						.setAckMode(ackMode);
 			}
 		}
+
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug("Listened partitions: "
 					+ StringUtils.collectionToCommaDelimitedString(listenedPartitions));

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -1639,7 +1639,7 @@ public class KafkaBinderTests extends
 				moduleOutputChannel, createProducerProperties());
 
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
-		consumerProperties.getExtension().setAutoCommitOffset(false);
+		consumerProperties.getExtension().setAckMode(ContainerProperties.AckMode.MANUAL);
 
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
 				"testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff", "test",
@@ -1737,7 +1737,7 @@ public class KafkaBinderTests extends
 		QueueChannel inbound1 = new QueueChannel();
 		ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties = createConsumerProperties();
 		consumerProperties.getExtension().setAutoRebalanceEnabled(false);
-		consumerProperties.getExtension().setAckEachRecord(true);
+		consumerProperties.getExtension().setAckMode(ContainerProperties.AckMode.RECORD);
 		Binding<MessageChannel> consumerBinding1 = binder.bindConsumer(testDestination,
 				"test1", inbound1, consumerProperties);
 		QueueChannel inbound2 = new QueueChannel();