The Retry API cleanup

* Improve tests performance using `delay(Duration.ZERO)` where we are not interested in the delay between attempts
* Make `StatefulRetryOperationsInterceptor` to use a `LinkedHashMap` with `removeEldestEntry` removal.
Essentially, keeping the cache clean from those stale entries.

Author: artembilan

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

@@ -16,10 +16,12 @@
 
 package org.springframework.amqp.rabbit.config;
 
+import java.time.Duration;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 
 import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
@@ -42,6 +44,7 @@
  *
  * @author Juergen Hoeller
  * @author Stephane Nicoll
+ * @author Artem Bilan
  *
  * @since 4.0
  */
@@ -57,15 +60,26 @@ public final class StatefulRetryOperationsInterceptor implements MethodIntercept
 
 	private final @Nullable MessageRecoverer messageRecoverer;
 
-	private final Map<Object, RetryState> cache = new ConcurrentHashMap<>();
+	private final int stateCacheSize;
+
+	private final Map<Object, RetryState> retryStateCache =
+			Collections.synchronizedMap(new LinkedHashMap<>(100, 0.75f, true) {
+
+				@Override
+				protected boolean removeEldestEntry(Map.Entry<Object, RetryState> eldest) {
+					return size() > StatefulRetryOperationsInterceptor.this.stateCacheSize;
+				}
+
+			});
 
 	StatefulRetryOperationsInterceptor(MessageKeyGenerator messageKeyGenerator,
 			NewMessageIdentifier newMessageIdentifier, @Nullable RetryPolicy retryPolicy,
-			@Nullable MessageRecoverer messageRecoverer) {
+			@Nullable MessageRecoverer messageRecoverer, @Nullable Integer stateCacheSize) {
 
 		this.messageKeyGenerator = messageKeyGenerator;
 		this.newMessageIdentifier = newMessageIdentifier;
-		this.retryPolicy = (retryPolicy != null) ? retryPolicy : RetryPolicy.builder().build();
+		this.retryPolicy = retryPolicy != null ? retryPolicy : RetryPolicy.builder().delay(Duration.ZERO).build();
+		this.stateCacheSize = stateCacheSize != null ? stateCacheSize : 1000;
 		this.messageRecoverer = messageRecoverer;
 	}
 
@@ -77,20 +91,20 @@ public final class StatefulRetryOperationsInterceptor implements MethodIntercept
 		if (key == null) { // Means no retry for this message anymore.
 			return invocation.proceed();
 		}
-		RetryState retryState = this.cache.get(key);
+		RetryState retryState = this.retryStateCache.get(key);
 		if (retryState == null || this.newMessageIdentifier.isNew(message)) {
 			try {
 				return invocation.proceed();
 			}
 			catch (Throwable ex) {
-				this.cache.put(key, new RetryState(this.retryPolicy.getBackOff().start(), ex));
+				this.retryStateCache.put(key, new RetryState(this.retryPolicy.getBackOff().start(), ex));
 				throw ex;
 			}
 		}
 		else {
 			long time = retryState.backOffExecution().nextBackOff();
 			if (time == BackOffExecution.STOP || !this.retryPolicy.shouldRetry(retryState.lastException())) {
-				this.cache.remove(key);
+				this.retryStateCache.remove(key);
 				recover(args[1], retryState.lastException());
 				// This is actually a normal outcome. It means the recovery was successful, but we don't want to consume
 				// any more messages until the acks and commits are sent for this (problematic) message...
@@ -101,11 +115,11 @@ public final class StatefulRetryOperationsInterceptor implements MethodIntercept
 				Thread.sleep(time);
 				try {
 					Object result = invocation.proceed();
-					this.cache.remove(key);
+					this.retryStateCache.remove(key);
 					return result;
 				}
 				catch (Throwable ex) {
-					this.cache.put(key, new RetryState(retryState.backOffExecution(), ex));
+					this.retryStateCache.put(key, new RetryState(retryState.backOffExecution(), ex));
 					throw ex;
 				}
 			}

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

@@ -18,16 +18,17 @@
 
 import org.jspecify.annotations.Nullable;
 
+import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.retry.MessageKeyGenerator;
 import org.springframework.amqp.rabbit.retry.MessageRecoverer;
 import org.springframework.amqp.rabbit.retry.NewMessageIdentifier;
 import org.springframework.core.retry.RetryPolicy;
 
 /**
  * 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
+ * a large amount of control over the behavior of a container when a listener fails. To control the number of retry
  * 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
+ * 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
  * be recognised (hence the {@link MessageKeyGenerator} strategy), and when the retry attempts are exhausted it will be
@@ -44,9 +45,20 @@
  */
 public class StatefulRetryOperationsInterceptorFactoryBean extends AbstractRetryOperationsInterceptorFactoryBean {
 
-	private @Nullable MessageKeyGenerator messageKeyGenerator;
+	private MessageKeyGenerator messageKeyGenerator =
+			(message) -> {
+				MessageProperties messageProperties = message.getMessageProperties();
+				String messageId = messageProperties.getMessageId();
+				if (messageId == null && Boolean.TRUE.equals(messageProperties.isRedelivered())) {
+					messageProperties.setFinalRetryForMessageWithNoId(true);
+				}
+				return messageId;
+			};
 
-	private @Nullable NewMessageIdentifier newMessageIdentifier;
+	private NewMessageIdentifier newMessageIdentifier =
+			(message) -> Boolean.FALSE.equals(message.getMessageProperties().isRedelivered());
+
+	private @Nullable Integer stateCacheSize;
 
 	public void setMessageKeyGenerator(MessageKeyGenerator messageKeyGenerator) {
 		this.messageKeyGenerator = messageKeyGenerator;
@@ -56,30 +68,14 @@ public void setNewMessageIdentifier(NewMessageIdentifier newMessageIdentifier) {
 		this.newMessageIdentifier = newMessageIdentifier;
 	}
 
-	@Override
-	public StatefulRetryOperationsInterceptor getObject() {
-		return new StatefulRetryOperationsInterceptor(getMessageKeyGenerator(),
-				getNewMessageIdentifier(), getRetryPolicy(), getMessageRecoverer());
+	public void setStateCacheSize(int stateCacheSize) {
+		this.stateCacheSize = stateCacheSize;
 	}
 
-	private NewMessageIdentifier getNewMessageIdentifier() {
-		if (this.newMessageIdentifier != null) {
-			return this.newMessageIdentifier;
-		}
-		return (message) -> Boolean.FALSE.equals(message.getMessageProperties().isRedelivered());
-	}
-
-	private MessageKeyGenerator getMessageKeyGenerator() {
-		if (this.messageKeyGenerator != null) {
-			return this.messageKeyGenerator;
-		}
-		return (message) -> {
-			String messageId = message.getMessageProperties().getMessageId();
-			if (messageId == null && Boolean.TRUE.equals(message.getMessageProperties().isRedelivered())) {
-				message.getMessageProperties().setFinalRetryForMessageWithNoId(true);
-			}
-			return messageId;
-		};
+	@Override
+	public StatefulRetryOperationsInterceptor getObject() {
+		return new StatefulRetryOperationsInterceptor(this.messageKeyGenerator,
+				this.newMessageIdentifier, getRetryPolicy(), getMessageRecoverer(), this.stateCacheSize);
 	}
 
 	@Override

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

@@ -16,6 +16,7 @@
 
 package org.springframework.amqp.rabbit.config;
 
+import java.time.Duration;
 import java.util.function.BiFunction;
 
 import org.aopalliance.intercept.MethodInterceptor;
@@ -33,6 +34,7 @@
  *
  * @author Juergen Hoeller
  * @author Stephane Nicoll
+ * @author Artem Bilan
  *
  * @since 4.0
  */
@@ -44,7 +46,8 @@ public final class StatelessRetryOperationsInterceptor implements MethodIntercep
 
 	StatelessRetryOperationsInterceptor(@Nullable RetryPolicy retryPolicy,
 			@Nullable BiFunction<@Nullable Object[], Throwable, @Nullable Object> recoverer) {
-		this.retryOperations = new RetryTemplate((retryPolicy != null) ? retryPolicy : RetryPolicy.builder().build());
+		this.retryOperations =
+				new RetryTemplate(retryPolicy != null ? retryPolicy : RetryPolicy.builder().delay(Duration.ZERO).build());
 		this.recoverer = recoverer;
 	}