diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index bc58e7daf..6af99f560 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/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. 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 9032cfc3c..4f85acb93 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 @@ -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; } 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 47213f6a1..e209aa6f4 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 @@ -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)); diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 974f2adb3..4604b80e6 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/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 consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setAutoCommitOffset(false); + consumerProperties.getExtension().setAckMode(ContainerProperties.AckMode.MANUAL); Binding consumerBinding = binder.bindConsumer( "testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff", "test", @@ -1737,7 +1737,7 @@ public class KafkaBinderTests extends QueueChannel inbound1 = new QueueChannel(); ExtendedConsumerProperties consumerProperties = createConsumerProperties(); consumerProperties.getExtension().setAutoRebalanceEnabled(false); - consumerProperties.getExtension().setAckEachRecord(true); + consumerProperties.getExtension().setAckMode(ContainerProperties.AckMode.RECORD); Binding consumerBinding1 = binder.bindConsumer(testDestination, "test1", inbound1, consumerProperties); QueueChannel inbound2 = new QueueChannel();