From 394b8a66852c4feacdc0aac77786d93549f72a02 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 9 Feb 2021 17:05:18 -0500 Subject: [PATCH] Restore autoCommitOffset behavior Until autoCommitOffset is fully removed (it is deprecated already in 3.1), honor the old behavior if the application uses the property. Restore autoCommitOnError from deprecated status to active for polled consumers. Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1026 --- docs/src/main/asciidoc/overview.adoc | 13 +++++---- .../properties/KafkaConsumerProperties.java | 24 +++------------- .../kafka/KafkaMessageChannelBinder.java | 28 +++++++++++++++---- 3 files changed, 35 insertions(+), 30 deletions(-) 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 properties) { return properties.getExtension().getAutoCommitOnError() != null ? properties.getExtension().getAutoCommitOnError() - : properties.getExtension().isAutoCommitOffset() - && properties.getExtension().isEnableDlq(); + : false; } private TopicPartitionOffset[] getTopicPartitionOffsets(