From 0d09e8db6df603c7cb62b04118a1d9ed32ed6bbc Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 2 Feb 2018 10:20:08 -0500 Subject: [PATCH] Support stateful retry * Polishing - we don't need to keep RetryStates in a map - in this context it is just a holder for the message key. Also PR comments. * Doc Polishing. --- ...AbstractKafkaListenerContainerFactory.java | 21 ++- .../config/AbstractKafkaListenerEndpoint.java | 23 ++- .../RetryingMessageListenerAdapter.java | 33 +++- .../kafka/annotation/StatefulRetryTests.java | 149 ++++++++++++++++++ src/reference/asciidoc/kafka.adoc | 22 ++- src/reference/asciidoc/whats-new.adoc | 3 + 6 files changed, 244 insertions(+), 7 deletions(-) create mode 100644 spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index 7d341652..f083ed10 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-2018 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. @@ -67,6 +67,8 @@ public abstract class AbstractKafkaListenerContainerFactory recoveryCallback; + private Boolean statefulRetry; + private Boolean batchListener; private ApplicationEventPublisher applicationEventPublisher; @@ -143,6 +145,20 @@ public abstract class AbstractKafkaListenerContainerFactory private RecoveryCallback recoveryCallback; + private boolean statefulRetry; + private boolean batchListener; private KafkaTemplate replyTemplate; @@ -303,6 +305,23 @@ public abstract class AbstractKafkaListenerEndpoint this.recoveryCallback = recoveryCallback; } + protected boolean isStatefulRetry() { + return this.statefulRetry; + } + + /** + * When using a {@link RetryTemplate}, set to true to enable stateful retry. Use in + * conjunction with a + * {@link org.springframework.kafka.listener.SeekToCurrentErrorHandler} when retry can + * take excessive time; each failure goes back to the broker, to keep the Consumer + * alive. + * @param statefulRetry true to enable stateful retry. + * @since 2.1.3 + */ + public void setStatefulRetry(boolean statefulRetry) { + this.statefulRetry = statefulRetry; + } + @Override public String getClientIdPrefix() { return this.clientIdPrefix; @@ -356,7 +375,7 @@ public abstract class AbstractKafkaListenerEndpoint Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener"); if (this.retryTemplate != null) { messageListener = new RetryingMessageListenerAdapter<>((MessageListener) messageListener, - this.retryTemplate, this.recoveryCallback); + this.retryTemplate, this.recoveryCallback, this.statefulRetry); } if (this.recordFilterStrategy != null) { if (this.batchListener) { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java index c23bbe95..fdfb6034 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -23,6 +23,8 @@ import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListe import org.springframework.kafka.listener.MessageListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.retry.RecoveryCallback; +import org.springframework.retry.RetryState; +import org.springframework.retry.support.DefaultRetryState; import org.springframework.retry.support.RetryTemplate; import org.springframework.util.Assert; @@ -56,6 +58,8 @@ public class RetryingMessageListenerAdapter */ public static final String CONTEXT_RECORD = "record"; + private boolean stateful; + /** * Construct an instance with the provided template and delegate. The exception will * be thrown to the container after retries are exhausted. @@ -75,13 +79,38 @@ public class RetryingMessageListenerAdapter */ public RetryingMessageListenerAdapter(MessageListener messageListener, RetryTemplate retryTemplate, RecoveryCallback recoveryCallback) { + this(messageListener, retryTemplate, recoveryCallback, false); + } + + /** + * Construct an instance with the provided template, callback and delegate. When using + * stateful retry, the retry context key is a concatenated String + * {@code topic-partition-offset}. A + * {@link org.springframework.kafka.listener.SeekToCurrentErrorHandler} is required in + * the listener container because stateful retry will throw the exception to the + * container for each delivery attempt. + * @param messageListener the delegate listener. + * @param retryTemplate the template. + * @param recoveryCallback the recovery callback; if null, the exception will be + * thrown to the container after retries are exhausted. + * @param stateful true for stateful retry. + * @since 2.1.3 + */ + public RetryingMessageListenerAdapter(MessageListener messageListener, RetryTemplate retryTemplate, + RecoveryCallback recoveryCallback, boolean stateful) { + super(messageListener, retryTemplate, recoveryCallback); Assert.notNull(messageListener, "'messageListener' cannot be null"); + this.stateful = stateful; } @Override public void onMessage(final ConsumerRecord record, final Acknowledgment acknowledgment, final Consumer consumer) { + RetryState retryState = null; + if (this.stateful) { + retryState = new DefaultRetryState(record.topic() + "-" + record.partition() + "-" + record.offset()); + } getRetryTemplate().execute(context -> { context.setAttribute(CONTEXT_RECORD, record); switch (RetryingMessageListenerAdapter.this.delegateType) { @@ -103,7 +132,7 @@ public class RetryingMessageListenerAdapter } return null; }, - getRecoveryCallback()); + getRecoveryCallback(), retryState); } /* diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java new file mode 100644 index 00000000..452cecbc --- /dev/null +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java @@ -0,0 +1,149 @@ +/* + * Copyright 2018 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 + * + * http://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.kafka.annotation; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.KafkaListenerContainerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.MessageListenerContainer; +import org.springframework.kafka.listener.SeekToCurrentErrorHandler; +import org.springframework.kafka.test.rule.KafkaEmbedded; +import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 2.1.3 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class StatefulRetryTests { + + private static final String DEFAULT_TEST_GROUP_ID = "statefulRetry"; + + @ClassRule + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, "sr1"); + + @Autowired + private Config config; + + @Autowired + private KafkaTemplate template; + + @Test + public void testStatefulRetry() throws Exception { + this.template.send("sr1", "foo"); + assertThat(this.config.latch1.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.latch2.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.seekPerformed).isTrue(); + } + + @Configuration + @EnableKafka + public static class Config { + + private final CountDownLatch latch1 = new CountDownLatch(3); + + private final CountDownLatch latch2 = new CountDownLatch(1); + + private boolean seekPerformed; + + @Bean + public KafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + factory.setConsumerFactory(consumerFactory()); + factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler() { + + @Override + public void handle(Exception thrownException, List> records, + Consumer consumer, MessageListenerContainer container) { + Config.this.seekPerformed = true; + super.handle(thrownException, records, consumer, container); + } + + }); + factory.setStatefulRetry(true); + factory.setRetryTemplate(new RetryTemplate()); + factory.setRecoveryCallback(c -> { + this.latch2.countDown(); + return null; + }); + return factory; + } + + @Bean + public DefaultKafkaConsumerFactory consumerFactory() { + return new DefaultKafkaConsumerFactory<>(consumerConfigs()); + } + + @Bean + public Map consumerConfigs() { + Map consumerProps = + KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + return consumerProps; + } + + @Bean + public KafkaTemplate template() { + KafkaTemplate kafkaTemplate = new KafkaTemplate<>(producerFactory()); + return kafkaTemplate; + } + + @Bean + public ProducerFactory producerFactory() { + return new DefaultKafkaProducerFactory<>(producerConfigs()); + } + + @Bean + public Map producerConfigs() { + return KafkaTestUtils.producerProps(embeddedKafka); + } + + @KafkaListener(id = "retry", topics = "sr1", groupId = "sr1") + public void listen1(String in) { + this.latch1.countDown(); + throw new RuntimeException("retry"); + } + + } + +} diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index f57e71f4..fef6dcad 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1049,15 +1049,16 @@ When using `@KafkaListener`, set the `RecordFilterStrategy` (and optionally `ack In addition, a `FilteringBatchMessageListenerAdapter` is provided, for when using a batch <>. +[[retrying-deliveries]] ===== Retrying Deliveries If your listener throws an exception, the default behavior is to invoke the `ErrorHandler`, if configured, or logged otherwise. NOTE: Two error handler interfaces are provided `ErrorHandler` and `BatchErrorHandler`; the appropriate type must be configured to match the <>. -To retry deliveries, convenient listener adapters - `RetryingMessageListenerAdapter` and `RetryingAcknowledgingMessageListenerAdapter` are provided, depending on whether you are using a `MessageListener` or an `AcknowledgingMessageListener`. +To retry deliveries, a convenient listener adapter `RetryingMessageListenerAdapter` is provided. -These can be configured with a `RetryTemplate` and `RecoveryCallback` - see the https://github.com/spring-projects/spring-retry[spring-retry] +It can be configured with a `RetryTemplate` and `RecoveryCallback` - see the https://github.com/spring-projects/spring-retry[spring-retry] project for information about these components. If a recovery callback is not provided, the exception is thrown to the container after retries are exhausted. In that case, the `ErrorHandler` will be invoked, if configured, or logged otherwise. @@ -1073,6 +1074,23 @@ See its javadocs for more information. A retry adapter is not provided for any of the batch <> because the framework has no knowledge of where, in a batch, the failure occurred. Users wishing retry capabilities, when using a batch listener, are advised to use a `RetryTemplate` within the listener itself. +[[stateful-retry]] +===== Stateful Retry + +It is important to understand that the retry discussed above suspends the consumer thread (if a `BackOffPolicy` is used); there are no calls to `Consumer.poll()` during the retries. +Kafka has two properties to determine consumer health; the `session.timeout.ms` is used to determine if the consumer is active. +Since version `0.10.1.0` heartbeats are sent on a background thread so a slow consumer no longer affects that. +`max.poll.interval.ms` (default 5 minutes) is used to determine if a consumer appears to be hung (taking too long to process records from the last poll). +If the time between `poll()` s exceeds this, the broker will revoke the assigned partitions and perform a rebalance. +For lengthy retry sequences, with back off, this can easily happen. + +Since _version 2.1.3_, you can avoid this problem by using stateful retry in conjunction with a `SeekToCurrentErrorHandler`. +In this case, each delivery attempt will throw the exception back to the container and the error handler will re-seek the unprocessed offsets and the same message will be redelivered by the next `poll()`. +This avoids the problem of exceeding the `max.poll.interval.ms` property (as long as an individual delay between attempts does not exceed it). +So, when using an `ExponentialBackOffPolicy`, it's important to ensure that the `maxInterval` is rather less than the `max.poll.interval.ms` property. +To enable stateful retry, use the `RetryingMessageListenerAdapter` constructor that takes a `stateful` `boolean` argument (set it to `true`). +When configuring using the listener container factory (for `@KafkaListener` s), set the factory's `statefulRetry` property to `true`. + [[idle-containers]] ===== Detecting Idle and Non-Responsive Consumers diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 93d7aef4..0559cca7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -15,6 +15,9 @@ See <> for more information. Container Error handlers are now provided for both record and batch listeners that treat any exceptions thrown by the listener as fatal; they stop the container. See <> for more information. +==== Stateful Retry + +Starting with _version 2.1.3_, stateful retry can be configured; see <> for more information. ==== Client ID