Some clean up for Retry functionality

* Add `whats-new.adoc` section about Core Retry migration
* Remove `SendRetryContextAccessor` as out-of-use API
* Remove `MessageRecoveryCallback` in favor of `MessageRecoverer`
* Improve `ReplyFailureException` to carry a `replyTo` property
* Use `ReplyFailureException` together with a `replyTo` in the `AbstractAdaptableMessageListener`
when retry is exhausted
* Improve Javadoc of the `MessageKeyGenerator` to justify a null from the `getKey()` contract
* Fix typos in the Javadocs of the affected classes

Author: artembilan

spring-amqp/src/main/java/org/springframework/amqp/support/SendRetryContextAccessor.java

@@ -30,14 +30,15 @@
  * Convenient factory bean for creating a stateless retry interceptor for use in a
  * {@link StreamListenerContainer} when consuming native stream messages, giving you a
  * large amount of control over the behavior of a container when a listener fails. To
- * control the number of retry attempt or the backoff in between attempts, supply a
+ * control the number of retry attempts or the backoff in between attempts, supply a
  * customized {@link RetryOperations}. Stateless retry is appropriate if your listener can
  * be called repeatedly between failures with no side effects. The semantics of stateless
  * retry mean that a listener exception is not propagated to the container until the retry
  * attempts are exhausted. When the retry attempts are exhausted it can be processed using
  * a {@link StreamMessageRecoverer} if one is provided.
  *
  * @author Gary Russell
+ * @author Stephane Nicoll
  */
 public class StreamRetryOperationsInterceptorFactoryBean extends StatelessRetryOperationsInterceptorFactoryBean {
 

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/retry/StreamRetryOperationsInterceptorFactoryBean.java

@@ -74,6 +74,8 @@
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Stephane Nicoll
+ *
  * @since 2.4
  *
  */
@@ -147,6 +149,7 @@ private Map<String, Object> queueInfo(String queueName) throws URISyntaxExceptio
 				.accept(MediaType.APPLICATION_JSON)
 				.retrieve()
 				.bodyToMono(new ParameterizedTypeReference<Map<String, Object>>() {
+
 				})
 				.block(Duration.ofSeconds(10));
 	}
@@ -233,20 +236,13 @@ public void start() {
 				}
 
 				private void clean(Environment env) {
-					try {
-						env.deleteStream("test.stream.queue1");
-					}
-					catch (Exception e) {
-					}
-					try {
-						env.deleteStream("test.stream.queue2");
-					}
-					catch (Exception e) {
-					}
-					try {
-						env.deleteStream("stream.created.over.amqp");
-					}
-					catch (Exception e) {
+					String[] streamsToDelete = {"test.stream.queue1", "test.stream.queue2", "stream.created.over.amqp"};
+					for (String streamToDelete : streamsToDelete) {
+						try {
+							env.deleteStream(streamToDelete);
+						}
+						catch (Exception e) {
+						}
 					}
 				}
 
@@ -274,11 +270,10 @@ RabbitListenerContainerFactory<StreamListenerContainer> rabbitListenerContainerF
 		RabbitListenerContainerFactory<StreamListenerContainer> observableFactory(Environment env) {
 			StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
 			factory.setObservationEnabled(true);
-			factory.setConsumerCustomizer((id, builder) -> {
-				builder.name(id)
-						.offset(OffsetSpecification.first())
-						.manualTrackingStrategy();
-			});
+			factory.setConsumerCustomizer((id, builder) ->
+					builder.name(id)
+							.offset(OffsetSpecification.first())
+							.manualTrackingStrategy());
 			return factory;
 		}
 
@@ -328,11 +323,10 @@ RabbitListenerContainerFactory<StreamListenerContainer> nativeObsFactory(Environ
 			StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
 			factory.setNativeListener(true);
 			factory.setObservationEnabled(true);
-			factory.setConsumerCustomizer((id, builder) -> {
-				builder.name(id)
-						.offset(OffsetSpecification.first())
-						.manualTrackingStrategy();
-			});
+			factory.setConsumerCustomizer((id, builder) ->
+					builder.name(id)
+							.offset(OffsetSpecification.first())
+							.manualTrackingStrategy());
 			return factory;
 		}
 

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java

@@ -27,6 +27,7 @@
  * Convenient base class for interceptor factories.
  *
  * @author Dave Syer
+ * @author Stephane Nicoll
  *
  */
 public abstract class AbstractRetryOperationsInterceptorFactoryBean implements FactoryBean<Advice> {
@@ -51,8 +52,4 @@ public void setMessageRecoverer(MessageRecoverer messageRecoverer) {
 		return this.messageRecoverer;
 	}
 
-	public boolean isSingleton() {
-		return true;
-	}
-
 }

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRetryOperationsInterceptorFactoryBean.java

@@ -29,7 +29,7 @@
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener;
 import org.springframework.amqp.rabbit.listener.adapter.ReplyPostProcessor;
-import org.springframework.amqp.rabbit.retry.MessageRecoveryCallback;
+import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.utils.JavaUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
@@ -45,6 +45,7 @@
  * @author Gary Russell
  * @author Ngoc Nhan
  * @author Artem Bilan
+ * @author Stephane Nicoll
  *
  * @since 2.4
  *
@@ -58,7 +59,7 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
 
 	private @Nullable RetryTemplate retryTemplate;
 
-	private @Nullable MessageRecoveryCallback recoveryCallback;
+	private @Nullable MessageRecoverer recoveryCallback;
 
 	private Advice @Nullable [] adviceChain;
 
@@ -93,9 +94,9 @@ public void setDefaultRequeueRejected(Boolean requeueRejected) {
 	}
 
 	/**
-	 * Set post processors that will be applied before sending replies; added to each
+	 * Set post-processors that will be applied before sending replies; added to each
 	 * message listener adapter.
-	 * @param postProcessors the post processors.
+	 * @param postProcessors the post-processors.
 	 * @see AbstractAdaptableMessageListener#setBeforeSendReplyPostProcessors(MessagePostProcessor...)
 	 */
 	public void setBeforeSendReplyPostProcessors(MessagePostProcessor... postProcessors) {
@@ -108,30 +109,30 @@ public void setBeforeSendReplyPostProcessors(MessagePostProcessor... postProcess
 	 * Set a {@link RetryTemplate} to use when sending replies; added to each message
 	 * listener adapter.
 	 * @param retryTemplate the template.
-	 * @see #setReplyRecoveryCallback(MessageRecoveryCallback)
+	 * @see #setReplyRecoveryCallback(MessageRecoverer)
 	 * @see AbstractAdaptableMessageListener#setRetryTemplate(RetryTemplate)
 	 */
 	public void setRetryTemplate(RetryTemplate retryTemplate) {
 		this.retryTemplate = retryTemplate;
 	}
 
 	/**
-	 * Set a {@link MessageRecoveryCallback} to invoke when retries are exhausted. Added to
+	 * Set a {@link MessageRecoverer} to invoke when retries are exhausted. Added to
 	 * each message listener adapter. Only used if a {@link #setRetryTemplate(RetryTemplate)
 	 * retryTemplate} is provided.
 	 * @param recoveryCallback the recovery callback.
 	 * @see #setRetryTemplate(RetryTemplate)
-	 * @see AbstractAdaptableMessageListener#setRecoveryCallback(MessageRecoveryCallback)
+	 * @see AbstractAdaptableMessageListener#setRecoveryCallback(MessageRecoverer)
 	 */
-	public void setReplyRecoveryCallback(MessageRecoveryCallback recoveryCallback) {
+	public void setReplyRecoveryCallback(MessageRecoverer recoveryCallback) {
 		this.recoveryCallback = recoveryCallback;
 	}
 
 	/**
-	 * Set a function to provide a reply post processor; it will be used if there is no
+	 * Set a function to provide a reply post-processor; it will be used if there is no
 	 * replyPostProcessor on the rabbit listener annotation. The input parameter is the
 	 * listener id.
-	 * @param replyPostProcessorProvider the post processor.
+	 * @param replyPostProcessorProvider the post-processor.
 	 * @since 3.0
 	 */
 	public void setReplyPostProcessorProvider(

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/BaseRabbitListenerContainerFactory.java

@@ -62,6 +62,7 @@
  * @author James Carr
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Stephane Nicoll
  *
  * @since 1.3
  *
@@ -100,13 +101,14 @@ protected final B _this() { // NOSONAR starts with _
 	}
 
 	/**
-	 * Apply the retry policy - cannot be used if a custom retry template has been provided, or if retry
+	 * Apply the retry policy - cannot be used if a custom retry template has been provided or if the retry
 	 * policy has been customized already.
 	 * @param policy The policy.
 	 * @return this.
 	 */
 	public B retryPolicy(RetryPolicy policy) {
-		Assert.isTrue(!this.templateAltered, "cannot set the retry policy if max attempts or back off policy or options changed");
+		Assert.isTrue(!this.templateAltered,
+				"cannot set the retry policy if max attempts or back off policy or options changed");
 		this.retryPolicy = policy;
 		this.retryPolicySet = true;
 		this.templateAltered = true;
@@ -135,7 +137,7 @@ public B maxAttempts(int maxAttempts) {
 	}
 
 	/**
-	 * Apply the backoff options. Cannot be used if a custom retry operations, or back off policy has been set.
+	 * Apply the backoff options. Cannot be used if a custom retry operations or back off policy has been set.
 	 * @param initialInterval The initial interval.
 	 * @param multiplier The multiplier.
 	 * @param maxInterval The max interval.
@@ -185,8 +187,8 @@ public static final class StatefulRetryInterceptorBuilder
 		}
 
 		/**
-		 * Stateful retry requires messages to be identifiable. Default is to use the message id header; use a custom
-		 * implementation if the message id is not present or not reliable.
+		 * Stateful retry requires messages to be identifiable. The default is to use the message id header;
+		 * use a custom implementation if the message id is not present or not reliable.
 		 * @param messageKeyGenerator The key generator.
 		 * @return this.
 		 */
@@ -196,7 +198,7 @@ public StatefulRetryInterceptorBuilder messageKeyGenerator(MessageKeyGenerator m
 		}
 
 		/**
-		 * Apply a custom new message identifier. Default is to use the redelivered header.
+		 * Apply a custom new message identifier. The default is to use the redelivered header.
 		 * @param newMessageIdentifier The new message identifier.
 		 * @return this.
 		 */

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/RetryInterceptorBuilder.java

@@ -42,6 +42,8 @@
  *
  * @author Juergen Hoeller
  * @author Stephane Nicoll
+ *
+ * @since 4.0
  */
 public final class StatefulRetryOperationsInterceptor implements MethodInterceptor {
 
@@ -60,6 +62,7 @@ public final class StatefulRetryOperationsInterceptor implements MethodIntercept
 	StatefulRetryOperationsInterceptor(MessageKeyGenerator messageKeyGenerator,
 			NewMessageIdentifier newMessageIdentifier, @Nullable RetryPolicy retryPolicy,
 			@Nullable MessageRecoverer messageRecoverer) {
+
 		this.messageKeyGenerator = messageKeyGenerator;
 		this.newMessageIdentifier = newMessageIdentifier;
 		this.retryPolicy = (retryPolicy != null) ? retryPolicy : RetryPolicy.builder().build();
@@ -71,7 +74,7 @@ public final class StatefulRetryOperationsInterceptor implements MethodIntercept
 		@Nullable Object[] args = invocation.getArguments();
 		Message message = argToMessage(args);
 		Object key = this.messageKeyGenerator.getKey(message);
-		if (key == null) { // state cannot be managed without a key
+		if (key == null) { // Means no retry for this message anymore.
 			return invocation.proceed();
 		}
 		RetryState retryState = this.cache.get(key);

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/StatefulRetryOperationsInterceptor.java

@@ -26,18 +26,19 @@
 /**
  * Convenient factory bean for creating a stateful retry interceptor for use in a message listener container, giving you
  * a large amount of control over the behaviour of a container when a listener fails. To control the number of retry
- * attempt or the backoff in between attempts, supply a customized {@link RetryPolicy}. Stateful retry is appropriate
- * if your listener is using a transactional resource that needs to be rollback on an exception (e.g. a stateful
+ * attempts or the backoff in between attempts, supply a customized {@link RetryPolicy}. Stateful retry is appropriate
+ * if your listener is using a transactional resource that needs to be rolled back on an exception (e.g. a stateful
  * connection to a back end server). JPA is the canonical example. The semantics of stateful retry mean that a listener
- * exception is propagated to the container, so that it can force a rollback. When the message is redelivered it has to
+ * exception is propagated to the container so that it can force a rollback. When the message is redelivered it has to
  * be recognised (hence the {@link MessageKeyGenerator} strategy), and when the retry attempts are exhausted it will be
- * processed using a {@link MessageRecoverer} if one is provided, in a new transaction. If a recoverer is not provided
+ * processed using a {@link MessageRecoverer} if one is provided, in a new transaction. If a recoverer is not provided,
  * the message will be logged and dropped.
  *
  * @author Dave Syer
  * @author Gary Russell
  * @author Ngoc Nhan
  * @author Artem Bilan
+ * @author Stephane Nicoll
  *
  * @see RetryPolicy#shouldRetry(Throwable)
  */

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/StatefulRetryOperationsInterceptorFactoryBean.java

@@ -33,6 +33,8 @@
  *
  * @author Juergen Hoeller
  * @author Stephane Nicoll
+ *
+ * @since 4.0
  */
 public final class StatelessRetryOperationsInterceptor implements MethodInterceptor {
 

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/StatelessRetryOperationsInterceptor.java

@@ -32,7 +32,7 @@
 /**
  * Convenient factory bean for creating a stateless retry interceptor for use in a message listener container, giving
  * you a large amount of control over the behaviour of a container when a listener fails. To control the number of retry
- * attempt or the backoff in between attempts, supply a customized {@link RetryPolicy}. Stateless retry is appropriate
+ * attempts or the backoff in between attempts, supply a customized {@link RetryPolicy}. Stateless retry is appropriate
  * if your listener can be called repeatedly between failures with no side effects. The semantics of stateless retry
  * mean that a listener exception is not propagated to the container until the retry attempts are exhausted. When the
  * retry attempts are exhausted it can be processed using a {@link MessageRecoverer} if one is provided, in the same
@@ -41,6 +41,7 @@
  *
  * @author Dave Syer
  * @author Gary Russell
+ * @author Stephane Nicoll
  *
  * @see RetryOperations#execute(Retryable)
  */