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.

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<C extends AbstractMe
 
 	private RecoveryCallback<? extends Object> recoveryCallback;
 
+	private Boolean statefulRetry;
+
 	private Boolean batchListener;
 
 	private ApplicationEventPublisher applicationEventPublisher;
@@ -143,6 +145,20 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 		this.recoveryCallback = recoveryCallback;
 	}
 
+	/**
+	 * 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;
+	}
+
+
 	/**
 	 * Return true if this endpoint creates a batch listener.
 	 * @return true for a batch listener.
@@ -216,6 +232,9 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 			if (this.recoveryCallback != null) {
 				aklEndpoint.setRecoveryCallback(this.recoveryCallback);
 			}
+			if (this.statefulRetry != null) {
+				aklEndpoint.setStatefulRetry(this.statefulRetry);
+			}
 			if (this.batchListener != null) {
 				aklEndpoint.setBatchListener(this.batchListener);
 			}

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.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.
@@ -88,6 +88,8 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private RecoveryCallback<? extends Object> recoveryCallback;
 
+	private boolean statefulRetry;
+
 	private boolean batchListener;
 
 	private KafkaTemplate<K, V> replyTemplate;
@@ -303,6 +305,23 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 		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<K, V>
 		Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener");
 		if (this.retryTemplate != null) {
 			messageListener = new RetryingMessageListenerAdapter<>((MessageListener<K, V>) messageListener,
-					this.retryTemplate, this.recoveryCallback);
+					this.retryTemplate, this.recoveryCallback, this.statefulRetry);
 		}
 		if (this.recordFilterStrategy != null) {
 			if (this.batchListener) {

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<K, V>
 	 */
 	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<K, V>
 	 */
 	public RetryingMessageListenerAdapter(MessageListener<K, V> messageListener, RetryTemplate retryTemplate,
 			RecoveryCallback<? extends Object> 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<K, V> messageListener, RetryTemplate retryTemplate,
+			RecoveryCallback<? extends Object> recoveryCallback, boolean stateful) {
+
 		super(messageListener, retryTemplate, recoveryCallback);
 		Assert.notNull(messageListener, "'messageListener' cannot be null");
+		this.stateful = stateful;
 	}
 
 	@Override
 	public void onMessage(final ConsumerRecord<K, V> 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<K, V>
 					}
 					return null;
 				},
-				getRecoveryCallback());
+				getRecoveryCallback(), retryState);
 	}
 
 	/*

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<Integer, String> 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<Integer, String> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(consumerFactory());
+			factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler() {
+
+				@Override
+				public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> 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<Integer, String> consumerFactory() {
+			return new DefaultKafkaConsumerFactory<>(consumerConfigs());
+		}
+
+		@Bean
+		public Map<String, Object> consumerConfigs() {
+			Map<String, Object> consumerProps =
+					KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", embeddedKafka);
+			consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+			return consumerProps;
+		}
+
+		@Bean
+		public KafkaTemplate<Integer, String> template() {
+			KafkaTemplate<Integer, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
+			return kafkaTemplate;
+		}
+
+		@Bean
+		public ProducerFactory<Integer, String> producerFactory() {
+			return new DefaultKafkaProducerFactory<>(producerConfigs());
+		}
+
+		@Bean
+		public Map<String, Object> producerConfigs() {
+			return KafkaTestUtils.producerProps(embeddedKafka);
+		}
+
+		@KafkaListener(id = "retry", topics = "sr1", groupId = "sr1")
+		public void listen1(String in) {
+			this.latch1.countDown();
+			throw new RuntimeException("retry");
+		}
+
+	}
+
+}

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 <<message-listeners, message listener>>.
 
+[[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 <<message-listeners, Message Listener>>.
 
-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<Void>` - see the https://github.com/spring-projects/spring-retry[spring-retry]
+It can be configured with a `RetryTemplate` and `RecoveryCallback<Void>` - 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 <<message-listeners, message listeners>> 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
 

src/reference/asciidoc/whats-new.adoc

@@ -15,6 +15,9 @@ See <<serdes>> 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 <<annotation-error-handling>> for more information.
 
+==== Stateful Retry
+
+Starting with _version 2.1.3_, stateful retry can be configured; see <<stateful-retry>> for more information.
 
 ==== Client ID