diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 55e6d401e..317716792 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -855,6 +855,88 @@ public interface KafkaBindingRebalanceListener { You cannot set the `resetOffsets` consumer property to `true` when you provide a rebalance listener. +[[retry-and-dlq-processing]] +=== Retry and Dead Letter Processing + +By default, when you configure retry (e.g. `maxAttemts`) and `enableDlq` in a consumer binding, these functions are performed within the binder, with no participation by the listener container or Kafka consumer. + +There are situations where it is preferable to move this functionality to the listener container, such as: + +* The aggregate of retries and delays will exceed the consumer's `max.poll.interval.ms` property, potentially causing a partition rebalance. +* You wish to publish the dead letter to a different Kafka cluster. +* You wish to add retry listeners to the error handler. +* ... + +To configure moving this functionality from the binder to the container, define a `@Bean` of type `ListenerContainerWithDlqAndRetryCustomizer`. +This interface has the following methods: + +==== +[source, java] +---- +/** + * Configure the container. + * @param container the container. + * @param destinationName the destination name. + * @param group the group. + * @param dlqDestinationResolver a destination resolver for the dead letter topic (if + * enableDlq). + * @param backOff the backOff using retry properties (if configured). + * @see #retryAndDlqInBinding(String, String) + */ +void configure(AbstractMessageListenerContainer container, String destinationName, String group, + @Nullable BiFunction, Exception, TopicPartition> dlqDestinationResolver, + @Nullable BackOff backOff); + +/** + * Return false to move retries and DLQ from the binding to a customized error handler + * using the retry metadata and/or a {@code DeadLetterPublishingRecoverer} when + * configured via + * {@link #configure(AbstractMessageListenerContainer, String, String, BiFunction, BackOff)}. + * @param destinationName the destination name. + * @param group the group. + * @return true to disable retrie in the binding + */ +default boolean retryAndDlqInBinding(String destinationName, String group) { + return true; +} +---- +==== + +The destination resolver and `BackOff` are created from the binding properties (if configured). +You can then use these to create a custom error handler and dead letter publisher; for example: + +==== +[source, java] +---- +@Bean +ListenerContainerWithDlqAndRetryCustomizer cust(KafkaTemplate template) { + return new ListenerContainerWithDlqAndRetryCustomizer() { + + @Override + public void configure(AbstractMessageListenerContainer container, String destinationName, + String group, + @Nullable BiFunction, Exception, TopicPartition> dlqDestinationResolver, + @Nullable BackOff backOff) { + + if (destinationName.equals("topicWithLongTotalRetryConfig")) { + ConsumerRecordRecoverer dlpr = new DeadLetterPublishingRecoverer(template), + dlqDestinationResolver); + container.setCommonErrorHandler(new DefaultErrorHandler(dlpr, backOff)); + } + } + + @Override + public boolean retryAndDlqInBinding(String destinationName, String group) { + return !destinationName.contains("topicWithLongTotalRetryConfig"); + } + + }; +} +---- +==== + +Now, only a single retry delay needs to be greater than the consumer's `max.poll.interval.ms` property. + [[consumer-producer-config-customizer]] === Customizing Consumer and Producer configuration 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 be2526dc3..5334e2eff 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 @@ -32,6 +32,7 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiFunction; import java.util.function.Predicate; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -110,6 +111,7 @@ import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.DefaultAfterRollbackProcessor; import org.springframework.kafka.listener.DefaultErrorHandler; import org.springframework.kafka.support.Acknowledgment; +import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries; import org.springframework.kafka.support.KafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.ProducerListener; @@ -733,13 +735,19 @@ public class KafkaMessageChannelBinder extends ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, consumerGroup, extendedConsumerProperties); + ListenerContainerCustomizer customizer = getContainerCustomizer(); + if (!extendedConsumerProperties.isBatchMode() && extendedConsumerProperties.getMaxAttempts() > 1 && transMan == null) { - kafkaMessageDrivenChannelAdapter - .setRetryTemplate(buildRetryTemplate(extendedConsumerProperties)); - kafkaMessageDrivenChannelAdapter - .setRecoveryCallback(errorInfrastructure.getRecoverer()); + if (!(customizer instanceof ListenerContainerWithDlqAndRetryCustomizer) + || ((ListenerContainerWithDlqAndRetryCustomizer) customizer) + .retryAndDlqInBinding(destination.getName(), group)) { + kafkaMessageDrivenChannelAdapter + .setRetryTemplate(buildRetryTemplate(extendedConsumerProperties)); + kafkaMessageDrivenChannelAdapter + .setRecoveryCallback(errorInfrastructure.getRecoverer()); + } if (!extendedConsumerProperties.getExtension().isEnableDlq()) { messageListenerContainer.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(0L, 0L))); } @@ -781,17 +789,43 @@ public class KafkaMessageChannelBinder extends CommonErrorHandler.class); messageListenerContainer.setCommonErrorHandler(commonErrorHandler); } - this.getContainerCustomizer().configure(messageListenerContainer, destination.getName(), group); + if (customizer instanceof ListenerContainerWithDlqAndRetryCustomizer) { + + BiFunction, Exception, TopicPartition> destinationResolver = createDestResolver( + extendedConsumerProperties.getExtension()); + BackOff createBackOff = extendedConsumerProperties.getMaxAttempts() > 1 + ? createBackOff(extendedConsumerProperties) + : null; + ((ListenerContainerWithDlqAndRetryCustomizer) customizer) + .configure(messageListenerContainer, destination.getName(), consumerGroup, destinationResolver, + createBackOff); + } + else { + ((ListenerContainerCustomizer) customizer) + .configure(messageListenerContainer, destination.getName(), consumerGroup); + } this.ackModeInfo.put(destination, messageListenerContainer.getContainerProperties().getAckMode()); return kafkaMessageDrivenChannelAdapter; } + private BiFunction, Exception, TopicPartition> createDestResolver( + KafkaConsumerProperties extension) { + + Integer dlqPartitions = extension.getDlqPartitions(); + if (extension.isEnableDlq()) { + return (rec, ex) -> dlqPartitions == null || dlqPartitions > 1 + ? new TopicPartition(extension.getDlqName(), rec.partition()) + : new TopicPartition(extension.getDlqName(), 0); + } + else { + return null; + } + } + /** * Configure a {@link BackOff} for the after rollback processor, based on the consumer * retry properties. If retry is disabled, return a {@link BackOff} that disables - * retry. Otherwise calculate the {@link ExponentialBackOff#setMaxElapsedTime(long)} - * so that the {@link BackOff} stops after the configured - * {@link ExtendedConsumerProperties#getMaxAttempts()}. + * retry. Otherwise use an {@link ExponentialBackOffWithMaxRetries}. * @param extendedConsumerProperties the properties. * @return the backoff. */ @@ -803,20 +837,10 @@ public class KafkaMessageChannelBinder extends return new FixedBackOff(0L, 0L); } int initialInterval = extendedConsumerProperties.getBackOffInitialInterval(); - double multiplier = extendedConsumerProperties.getBackOffMultiplier(); int maxInterval = extendedConsumerProperties.getBackOffMaxInterval(); - ExponentialBackOff backOff = new ExponentialBackOff(initialInterval, multiplier); + ExponentialBackOff backOff = new ExponentialBackOffWithMaxRetries(maxAttempts - 1); + backOff.setInitialInterval(initialInterval); backOff.setMaxInterval(maxInterval); - long maxElapsed = extendedConsumerProperties.getBackOffInitialInterval(); - double accum = maxElapsed; - for (int i = 1; i < maxAttempts - 1; i++) { - accum = accum * multiplier; - if (accum > maxInterval) { - accum = maxInterval; - } - maxElapsed += accum; - } - backOff.setMaxElapsedTime(maxElapsed); return backOff; } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/ListenerContainerWithDlqAndRetryCustomizer.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/ListenerContainerWithDlqAndRetryCustomizer.java new file mode 100644 index 000000000..567ac6441 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/ListenerContainerWithDlqAndRetryCustomizer.java @@ -0,0 +1,71 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka; + +import java.util.function.BiFunction; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; + +import org.springframework.cloud.stream.config.ListenerContainerCustomizer; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.lang.Nullable; +import org.springframework.util.backoff.BackOff; + +/** + * An extension of {@link ListenerContainerCustomizer} that provides access to dead letter + * metadata. + * + * @author Gary Russell + * @since 3.2 + * + */ +public interface ListenerContainerWithDlqAndRetryCustomizer + extends ListenerContainerCustomizer> { + + @Override + default void configure(AbstractMessageListenerContainer container, String destinationName, String group) { + } + + /** + * Configure the container. + * @param container the container. + * @param destinationName the destination name. + * @param group the group. + * @param dlqDestinationResolver a destination resolver for the dead letter topic (if + * enableDlq). + * @param backOff the backOff using retry properties (if configured). + * @see #retryAndDlqInBinding(String, String) + */ + void configure(AbstractMessageListenerContainer container, String destinationName, String group, + @Nullable BiFunction, Exception, TopicPartition> dlqDestinationResolver, + @Nullable BackOff backOff); + + /** + * Return false to move retries and DLQ from the binding to a customized error handler + * using the retry metadata and/or a {@code DeadLetterPublishingRecoverer} when + * configured via + * {@link #configure(AbstractMessageListenerContainer, String, String, BiFunction, BackOff)}. + * @param destinationName the destination name. + * @param group the group. + * @return true to disable retrie in the binding + */ + default boolean retryAndDlqInBinding(String destinationName, String group) { + return true; + } + +} diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java new file mode 100644 index 000000000..2002ee773 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java @@ -0,0 +1,117 @@ +/* + * Copyright 2021-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder.kafka.integration; + +import java.util.function.BiFunction; +import java.util.function.Consumer; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.kafka.ListenerContainerWithDlqAndRetryCustomizer; +import org.springframework.cloud.stream.binding.BindingsLifecycleController; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.core.KafkaOperations; +import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.CommonErrorHandler; +import org.springframework.kafka.listener.ConsumerRecordRecoverer; +import org.springframework.kafka.listener.DeadLetterPublishingRecoverer; +import org.springframework.kafka.listener.DefaultErrorHandler; +import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries; +import org.springframework.kafka.test.context.EmbeddedKafka; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.lang.Nullable; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.util.backoff.BackOff; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +/** + * @author Gary Russell + */ +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { + "spring.cloud.function.definition=retryInBinder;retryInContainer", + "spring.cloud.stream.bindings.retryInBinder-in-0.group=foo", + "spring.cloud.stream.bindings.retryInContainer-in-0.group=bar", + "spring.cloud.stream.kafka.bindings.retryInBinder-in-0.consumer.enable-dlq=true", + "spring.cloud.stream.kafka.bindings.retryInContainer-in-0.consumer.enable-dlq=true"}) +@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") +@DirtiesContext +public class KafkaRetryDlqBinderOrContainerTests { + + @Test + public void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) { + Binding retryInBinder = controller.queryState("retryInBinder-in-0"); + assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, "lifecycle.retryTemplate")).isNotNull(); + assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, + "lifecycle.messageListenerContainer.commonErrorHandler")).isNull(); + Binding retryInContainer = controller.queryState("retryInContainer-in-0"); + assertThat(KafkaTestUtils.getPropertyValue(retryInContainer, "lifecycle.retryTemplate")).isNull(); + assertThat(KafkaTestUtils.getPropertyValue(retryInContainer, + "lifecycle.messageListenerContainer.commonErrorHandler")).isInstanceOf(CommonErrorHandler.class); + assertThat(KafkaTestUtils.getPropertyValue(retryInContainer, + "lifecycle.messageListenerContainer.commonErrorHandler.failureTracker.backOff")) + .isInstanceOf(ExponentialBackOffWithMaxRetries.class); + } + + @SpringBootApplication + public static class ConfigCustomizerTestConfig { + + @Bean + public Consumer retryInBinder() { + return str -> { }; + } + + @Bean + public Consumer retryInContainer() { + return str -> { }; + } + + @Bean + ListenerContainerWithDlqAndRetryCustomizer cust() { + return new ListenerContainerWithDlqAndRetryCustomizer() { + + @Override + public void configure(AbstractMessageListenerContainer container, String destinationName, + String group, + BiFunction, Exception, TopicPartition> dlqDestinationResolver, + @Nullable BackOff backOff) { + + if (destinationName.contains("Container")) { + ConsumerRecordRecoverer dlpr = new DeadLetterPublishingRecoverer(mock(KafkaOperations.class), + dlqDestinationResolver); + container.setCommonErrorHandler(new DefaultErrorHandler(dlpr, backOff)); + } + } + + @Override + public boolean retryAndDlqInBinding(String destinationName, String group) { + return !destinationName.contains("Container"); + } + + }; + } + + } + +}