SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Restore autoCommitOffset behavior

Browse Source 
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
This commit is contained in:
Soby Chacko
2021-02-09 17:05:18 -05:00
committed by Gary Russell
parent 32939e30fe
commit 394b8a6685
3 changed files with 35 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
docs/src/main/asciidoc/overview.adoc
View File

@@ -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::

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

@@ -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;
}

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

@@ -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(