GH-1886: Deprecate RetryTemplate

Resolves #1886

* Missed a deprecation warning.

* Use FQCN.

Author: garyrussell

Diff for: spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -2310,99 +2310,7 @@ IMPORTANT: The `FilteringBatchMessageListenerAdapter` is ignored if your `@Kafka
 [[retrying-deliveries]]
 ===== Retrying Deliveries
 
-If your listener throws an exception, the default behavior is to invoke the <<error-handlers>>, if configured, or logged otherwise.
-
-NOTE:
-To retry deliveries, a convenient listener adapter `RetryingMessageListenerAdapter`  is provided.
-
-You can configure it 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` is invoked, if configured, or logged otherwise.
-
-When you use `@KafkaListener`, you can set the `RetryTemplate` (and optionally `recoveryCallback`) on the container factory.
-When you do so, the listener is wrapped in the appropriate retrying adapter.
-
-The contents of the `RetryContext` passed into the `RecoveryCallback` depend on the type of listener.
-The context always has a `record` attribute, which is the record for which the failure occurred.
-If your listener is acknowledging or consumer aware, additional `acknowledgment` or `consumer` attributes are available.
-For convenience, the `RetryingMessageListenerAdapter` provides static constants for these keys.
-See its https://docs.spring.io/spring-kafka/api/org/springframework/kafka/listener/adapter/AbstractRetryingMessageListenerAdapter.html[Javadoc] 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.
-If you need retry capabilities when you use a batch listener, we recommend that you use a `RetryTemplate` within the listener itself.
-
-[[stateful-retry]]
-===== Stateful Retry
-
-IMPORTANT: Now that the `SeekToCurrentErrorHandler` can be configured with a `BackOff` and has the ability to retry only certain exceptions (since version 2.3), the use of stateful retry, via the listener adapter retry configuration, is no longer necessary.
-You can provide the same functionality with appropriate configuration of the error handler and remove all retry configuration from the listener adapter.
-See <<seek-to-current>> for more information.
-
-You should understand that the retry discussed in the <<retrying-deliveries,preceding section>> 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 `kafka-clients` 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: five 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()` calls exceeds this, the broker revokes the assigned partitions and performs 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 throws the exception back to the container, the error handler re-seeks the unprocessed offsets, and the same message is 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 you use an `ExponentialBackOffPolicy`, you must ensure that the `maxInterval` is less than the `max.poll.interval.ms` property.
-To enable stateful retry, you can use the `RetryingMessageListenerAdapter` constructor that takes a `stateful` `boolean` argument (set it to `true`).
-When you configure the listener container factory (for `@KafkaListener`), set the factory's `statefulRetry` property to `true`.
-
-IMPORTANT: Version 2.2 added recovery to the `SeekToCurrentErrorHandler`, such as sending a failed record to a dead-letter topic.
-When using stateful retry, you must perform the recovery in the retry `RecoveryCallback` and NOT in the error handler.
-Otherwise, if the recovery is done in the error handler, the retry template's state will never be cleared.
-Also, you must ensure that the `maxFailures` in the `SeekToCurrentErrorHandler` must be at least as many as configured in the retry policy, again to ensure that the retries are exhausted and the state cleared.
-Here is an example for retry configuration when used with a `SeekToCurrentErrorHandler` where `factory` is the `ConcurrentKafkaListenerContainerFactory`.
-
-====
-[source, java]
-----
-@Autowired
-DeadLetterPublishingRecoverer recoverer;
-
-...
-    factory.setRetryTemplate(new RetryTemplate()); // 3 retries by default
-    factory.setStatefulRetry(true);
-    factory.setRecoveryCallback(context -> {
-        recoverer.accept((ConsumerRecord<?, ?>) context.getAttribute("record"),
-                (Exception) context.getLastThrowable());
-        return null;
-    });
-...
-
-@Bean
-public SeekToCurrentErrorHandler eh() {
-    return new SeekToCurrentErrorHandler(new FixedBackOff(0L, 3L)); // at least 3
-}
-----
-====
-
-However, see the note at the beginning of this section; you can avoid using the `RetryTemplate` altogether.
-
-IMPORTANT: If the recoverer fails (throws an exception), the failed record will be included in the seeks.
-Starting with version 2.5.5, if the recoverer fails, the `BackOff` will be reset by default and redeliveries will again go through the back offs before recovery is attempted again.
-With earlier versions, the `BackOff` was not reset and recovery was re-attempted on the next failure.
-To revert to the previous behavior, set the error handler's `resetStateOnRecoveryFailure` to `false`.
-
-Starting with version 2.6, you can now provide the error handler with a `BiFunction<ConsumerRecord<?, ?>, Exception, BackOff>` to determine the `BackOff` to use, based on the failed record and/or the exception:
-
-====
-[source, java]
-----
-handler.setBackOffFunction((record, ex) -> { ... });
-----
-====
-
-If the function returns `null`, the handler's default `BackOff` will be used.
-
-Starting with version 2.6.3, set `resetStateOnExceptionChange` to `true` and the retry sequence will be restarted (including the selection of a new `BackOff`, if so configured) if the exception type changes between failures.
-By default, the exception type is not considered.
+See the `SeekToCurrentErrorHandler` in <<annotation-error-handling>>.
 
 [[sequencing]]
 ===== Starting `@KafkaListener` s in Sequence

Diff for: spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -180,7 +180,9 @@ public void setAckDiscarded(Boolean ackDiscarded) {
 	/**
 	 * Set a retryTemplate.
 	 * @param retryTemplate the template.
+	 * @deprecated since 2.8 - use a suitably configured error handler instead.
 	 */
+	@Deprecated
 	public void setRetryTemplate(RetryTemplate retryTemplate) {
 		this.retryTemplate = retryTemplate;
 	}
@@ -385,6 +387,7 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		return instance;
 	}
 
+	@SuppressWarnings("deprecation")
 	private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint) {
 		JavaUtils.INSTANCE
 				.acceptIfNotNull(this.recordFilterStrategy, aklEndpoint::setRecordFilterStrategy)

Diff for: spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2020 the original author or authors.
+ * Copyright 2014-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.
@@ -45,7 +45,6 @@
 import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.adapter.ReplyHeadersConfigurer;
-import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.lang.Nullable;
@@ -324,7 +323,9 @@ protected RetryTemplate getRetryTemplate() {
 	/**
 	 * Set a retryTemplate.
 	 * @param retryTemplate the template.
+	 * @deprecated since 2.8 - use a suitably configured error handler instead.
 	 */
+	@Deprecated
 	public void setRetryTemplate(RetryTemplate retryTemplate) {
 		this.retryTemplate = retryTemplate;
 	}
@@ -498,7 +499,7 @@ public void setupListenerContainer(MessageListenerContainer listenerContainer,
 	protected abstract MessagingMessageListenerAdapter<K, V> createMessageListener(MessageListenerContainer container,
 			@Nullable MessageConverter messageConverter);
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "deprecation" })
 	private void setupMessageListener(MessageListenerContainer container,
 			@Nullable MessageConverter messageConverter) {
 
@@ -514,7 +515,8 @@ private void setupMessageListener(MessageListenerContainer container,
 				"A 'RetryTemplate' is not supported with a batch listener; consider configuring the container "
 				+ "with a suitably configured 'SeekToCurrentBatchErrorHandler' instead");
 		if (this.retryTemplate != null) {
-			messageListener = new RetryingMessageListenerAdapter<>((MessageListener<K, V>) messageListener,
+			messageListener = new org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter<>(
+							(MessageListener<K, V>) messageListener,
 					this.retryTemplate, this.recoveryCallback, this.statefulRetry);
 		}
 		if (this.recordFilterStrategy != null) {

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapter.java

@@ -35,9 +35,12 @@
  * @param <K> the key type.
  * @param <V> the value type.
  *
+ * @deprecated since 2.8 - use a suitably configured error handler instead.
+ *
  * @author Gary Russell
  *
  */
+@Deprecated
 public class RetryingMessageListenerAdapter<K, V>
 		extends AbstractRetryingMessageListenerAdapter<K, V, MessageListener<K, V>>
 		implements AcknowledgingConsumerAwareMessageListener<K, V> {

Diff for: spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -111,7 +111,6 @@
 import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
-import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.KafkaNull;
@@ -255,6 +254,7 @@ public void testAnonymous() {
 		container.stop();
 	}
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testSimple() throws Exception {
 		this.recordFilter.called = false;
@@ -303,7 +303,7 @@ public void testSimple() throws Exception {
 		assertThat(KafkaTestUtils.getPropertyValue(manualContainer, "containerProperties.messageListener.ackDiscarded",
 				Boolean.class)).isTrue();
 		assertThat(KafkaTestUtils.getPropertyValue(manualContainer, "containerProperties.messageListener.delegate"))
-				.isInstanceOf(RetryingMessageListenerAdapter.class);
+				.isInstanceOf(org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter.class);
 		assertThat(KafkaTestUtils
 				.getPropertyValue(manualContainer, "containerProperties.messageListener.delegate.recoveryCallback")
 				.getClass().getName()).contains("EnableKafkaIntegrationTests$Config$");
@@ -1197,6 +1197,7 @@ public KafkaListenerContainerFactory<?> batchManualFactory2() {
 		}
 
 		@Bean
+		@SuppressWarnings("deprecation")
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
 				kafkaManualAckListenerContainerFactory() {
 

Diff for: spring-kafka/src/test/java/org/springframework/kafka/annotation/StatefulRetryTests.java

@@ -82,6 +82,7 @@ public static class Config {
 
 		private boolean seekPerformed;
 
+		@SuppressWarnings("deprecation")
 		@Bean
 		public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(EmbeddedKafkaBroker embeddedKafka) {
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =

Diff for: spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/RetryingMessageListenerAdapterTests.java

@@ -37,6 +37,7 @@
  * @since 2.0
  *
  */
+@SuppressWarnings("deprecation")
 public class RetryingMessageListenerAdapterTests {
 
 	@Test