diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc
index 2e870cfc9..606567858 100644
--- a/docs/src/main/asciidoc/overview.adoc
+++ b/docs/src/main/asciidoc/overview.adoc
@@ -192,13 +192,17 @@ 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::
+
+Starting with version 3.1, this property is deprecated.
+See `ackMode` for more details on alternatives.
 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.
 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`. This property is deprecated as of 3.1. See `ackMode` for more details.
+Also see `ackEachRecord`.
 +
 Default: `true`.
 ackMode::
@@ -207,10 +211,9 @@ 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.
-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.
+In pollable consumers, if set to `true`, it always auto commits on error.
+If not set (the default) or false, it will not auto commit in pollable consumers.
+Note that this property is only applicable for pollable consumers.
 +
 Default: not set.
 resetOffsets::
diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
index 4f85acb93..9b1087227 100644
--- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
+++ b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaConsumerProperties.java
@@ -114,14 +114,8 @@ public class KafkaConsumerProperties {
 	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.
-	 * It allows a stream to automatically replay from the last successfully processed message, in case of persistent failures.
-	 * 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.
+	 * Flag to enable auto commit on error in polled consumers.
 	 */
-	@Deprecated
 	private Boolean autoCommitOnError;
 
 	/**
@@ -312,29 +306,19 @@ public class KafkaConsumerProperties {
 	}
 
 	/**
-	 * @return is autocommit on error
+	 * @return is autocommit on error in polled consumers.
 	 *
-	 * 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.
-	 * 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.
+	 * This property accessor is only used in polled consumers.
 	 */
-	@Deprecated
 	public Boolean getAutoCommitOnError() {
 		return this.autoCommitOnError;
 	}
 
 	/**
 	 *
-	 * @param autoCommitOnError commit on error
+	 * @param autoCommitOnError commit on error in polled consumers.
 	 *
-	 * @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;
 	}
diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
index 979df4b5c..7858f5969 100644
--- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
+++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
@@ -694,10 +694,27 @@ public class KafkaMessageChannelBinder extends
 		messageListenerContainer.setBeanName(destination + ".container");
 		// end of these won't be needed...
 		ContainerProperties.AckMode ackMode = extendedConsumerProperties.getExtension().getAckMode();
-		if (ackMode == null && extendedConsumerProperties.getExtension().isAckEachRecord()) {
-			ackMode = ContainerProperties.AckMode.RECORD;
+		if (ackMode == null) {
+			if (extendedConsumerProperties.getExtension().isAckEachRecord()) {
+				ackMode = ContainerProperties.AckMode.RECORD;
+			}
+			else {
+				if (!extendedConsumerProperties.getExtension().isAutoCommitOffset()) {
+					messageListenerContainer.getContainerProperties()
+							.setAckMode(ContainerProperties.AckMode.MANUAL);
+					messageListenerContainer.getContainerProperties().setAckOnError(false);
+				}
+				else {
+					messageListenerContainer.getContainerProperties()
+							.setAckOnError(isAutoCommitOnError(extendedConsumerProperties));
+					if (extendedConsumerProperties.getExtension().isAckEachRecord()) {
+						messageListenerContainer.getContainerProperties()
+								.setAckMode(ContainerProperties.AckMode.RECORD);
+					}
+				}
+			}
 		}
-		if (ackMode != null) {
+		else {
 			if ((extendedConsumerProperties.isBatchMode() && ackMode != ContainerProperties.AckMode.RECORD) ||
 					!extendedConsumerProperties.isBatchMode()) {
 				messageListenerContainer.getContainerProperties()
@@ -705,6 +722,8 @@ public class KafkaMessageChannelBinder extends
 			}
 		}
 
+
+
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug("Listened partitions: "
 					+ StringUtils.collectionToCommaDelimitedString(listenedPartitions));
@@ -1397,8 +1416,7 @@ public class KafkaMessageChannelBinder extends
 			ExtendedConsumerProperties<KafkaConsumerProperties> properties) {
 		return properties.getExtension().getAutoCommitOnError() != null
 				? properties.getExtension().getAutoCommitOnError()
-				: properties.getExtension().isAutoCommitOffset()
-						&& properties.getExtension().isEnableDlq();
+				: false;
 	}
 
 	private TopicPartitionOffset[] getTopicPartitionOffsets(