Support timeouts in @​Retryable and RetryPolicy

Specifically, this commit introduces:

- timeout and timeoutString attributes in @​Retryable

- a default getTimeout() method in RetryPolicy

- a timeout() method in RetryPolicy.Builder

- an onRetryPolicyTimeout() callback in RetryListener

- support for checking exceeded timeouts in RetryTemplate (also used
  for imperative method invocations with @​Retryable)

- support for checking exceeded timeouts in reactive pipelines with
  @​Retryable

Closes gh-35963

Author: sbrannen

spring-context/src/main/java/org/springframework/resilience/annotation/RetryAnnotationBeanPostProcessor.java

@@ -99,6 +99,7 @@ private class RetryAnnotationInterceptor extends AbstractRetryInterceptor {
 					Arrays.asList(retryable.includes()), Arrays.asList(retryable.excludes()),
 					instantiatePredicate(retryable.predicate()),
 					parseLong(retryable.maxRetries(), retryable.maxRetriesString()),
+					parseDuration(retryable.timeout(), retryable.timeoutString(), timeUnit),
 					parseDuration(retryable.delay(), retryable.delayString(), timeUnit),
 					parseDuration(retryable.jitter(), retryable.jitterString(), timeUnit),
 					parseDouble(retryable.multiplier(), retryable.multiplierString()),

spring-context/src/main/java/org/springframework/resilience/annotation/Retryable.java

@@ -122,6 +122,39 @@
 	 */
 	String maxRetriesString() default "";
 
+	/**
+	 * The maximum amount of elapsed time allowed for the initial invocation and
+	 * any subsequent retry attempts, including delays.
+	 * <p>The default is {@code 0}, which signals that no timeout should be applied.
+	 * <p>The time unit is milliseconds by default but can be overridden via
+	 * {@link #timeUnit}.
+	 * <p>Must be greater than or equal to zero.
+	 * @since 7.0.2
+	 */
+	long timeout() default 0;
+
+	/**
+	 * The timeout, as a duration String.
+	 * <p>A non-empty value specified here overrides the {@link #timeout()} attribute.
+	 * <p>The duration String can be in several formats:
+	 * <ul>
+	 * <li>a plain integer &mdash; which is interpreted to represent a duration in
+	 * milliseconds by default unless overridden via {@link #timeUnit()} (prefer
+	 * using {@link #delay()} in that case)</li>
+	 * <li>any of the known {@link org.springframework.format.annotation.DurationFormat.Style
+	 * DurationFormat.Style}: the {@link org.springframework.format.annotation.DurationFormat.Style#ISO8601 ISO8601}
+	 * style or the {@link org.springframework.format.annotation.DurationFormat.Style#SIMPLE SIMPLE} style
+	 * &mdash; using the {@link #timeUnit()} as fallback if the string doesn't contain an explicit unit</li>
+	 * <li>one of the above, with Spring-style "${...}" placeholders as well as SpEL expressions</li>
+	 * </ul>
+	 * @return the timeout as a String value &mdash; for example, a placeholder, a
+	 * {@link org.springframework.format.annotation.DurationFormat.Style#ISO8601 java.time.Duration} compliant value,
+	 * or a {@link org.springframework.format.annotation.DurationFormat.Style#SIMPLE simple format} compliant value
+	 * @since 7.0.2
+	 * @see #timeout()
+	 */
+	String timeoutString() default "";
+
 	/**
 	 * The base delay after the initial invocation. If a multiplier is specified,
 	 * this serves as the initial delay to multiply from.

spring-context/src/main/java/org/springframework/resilience/retry/AbstractRetryInterceptor.java

@@ -17,6 +17,7 @@
 package org.springframework.resilience.retry;
 
 import java.lang.reflect.Method;
+import java.time.Duration;
 import java.util.concurrent.Future;
 
 import org.aopalliance.intercept.MethodInterceptor;
@@ -94,6 +95,7 @@ public AbstractRetryInterceptor() {
 				.excludes(spec.excludes())
 				.predicate(spec.predicate().forMethod(method))
 				.maxRetries(spec.maxRetries())
+				.timeout(spec.timeout())
 				.delay(spec.delay())
 				.jitter(spec.jitter())
 				.multiplier(spec.multiplier())
@@ -142,8 +144,20 @@ public static Object adaptReactiveResult(
 					.multiplier(spec.multiplier())
 					.maxBackoff(spec.maxDelay())
 					.filter(spec.combinedPredicate().forMethod(method));
-			publisher = (adapter.isMultiValue() ? Flux.from(publisher).retryWhen(retry) :
-					Mono.from(publisher).retryWhen(retry));
+
+			Duration timeout = spec.timeout();
+			boolean timeoutIsPositive = (!timeout.isNegative() && !timeout.isZero());
+			if (adapter.isMultiValue()) {
+				publisher = (timeoutIsPositive ?
+						Flux.from(publisher).retryWhen(retry).timeout(timeout) :
+						Flux.from(publisher).retryWhen(retry));
+			}
+			else {
+				publisher = (timeoutIsPositive ?
+						Mono.from(publisher).retryWhen(retry).timeout(timeout) :
+						Mono.from(publisher).retryWhen(retry));
+			}
+
 			return adapter.fromPublisher(publisher);
 		}
 

spring-context/src/main/java/org/springframework/resilience/retry/MethodRetrySpec.java

@@ -28,11 +28,14 @@
  * on {@link org.springframework.resilience.annotation.Retryable}.
  *
  * @author Juergen Hoeller
+ * @author Sam Brannen
  * @since 7.0
  * @param includes applicable exception types to attempt a retry for
  * @param excludes non-applicable exception types to avoid a retry for
  * @param predicate a predicate for filtering exceptions from applicable methods
  * @param maxRetries the maximum number of retry attempts
+ * @param timeout the maximum amount of elapsed time allowed for the initial
+ * invocation and any subsequent retry attempts, including delays
  * @param delay the base delay after the initial invocation
  * @param jitter a jitter value for the next retry attempt
  * @param multiplier a multiplier for a delay for the next retry attempt
@@ -46,20 +49,40 @@ public record MethodRetrySpec(
 		Collection<Class<? extends Throwable>> excludes,
 		MethodRetryPredicate predicate,
 		long maxRetries,
+		Duration timeout,
 		Duration delay,
 		Duration jitter,
 		double multiplier,
 		Duration maxDelay) {
 
+	/**
+	 * Construct a new {@code MethodRetryPredicate} with the supplied arguments.
+	 */
 	public MethodRetrySpec(MethodRetryPredicate predicate, long maxRetries, Duration delay) {
 		this(predicate, maxRetries, delay, Duration.ZERO, 1.0, Duration.ofMillis(Long.MAX_VALUE));
 	}
 
+	/**
+	 * Construct a new {@code MethodRetryPredicate} with the supplied arguments.
+	 */
 	public MethodRetrySpec(MethodRetryPredicate predicate, long maxRetries, Duration delay,
 			Duration jitter, double multiplier, Duration maxDelay) {
 
-		this(Collections.emptyList(), Collections.emptyList(), predicate, maxRetries, delay,
-				jitter, multiplier, maxDelay);
+		this(Collections.emptyList(), Collections.emptyList(), predicate, maxRetries, Duration.ZERO,
+				delay, jitter, multiplier, maxDelay);
+	}
+
+	/**
+	 * Construct a new {@code MethodRetryPredicate} with the supplied arguments.
+	 * @deprecated as of Spring Framework 7.0.2, in favor of
+	 * {@link #MethodRetrySpec(Collection, Collection, MethodRetryPredicate, long, Duration, Duration, Duration, double, Duration)}
+	 */
+	@Deprecated(since = "7.0.2", forRemoval = true)
+	public MethodRetrySpec(Collection<Class<? extends Throwable>> includes,
+			Collection<Class<? extends Throwable>> excludes, MethodRetryPredicate predicate,
+			long maxRetries, Duration delay, Duration jitter, double multiplier, Duration maxDelay) {
+
+		this(includes, excludes, predicate, maxRetries, Duration.ZERO, delay, jitter, multiplier, maxDelay);
 	}
 
 

spring-context/src/test/java/org/springframework/resilience/ReactiveRetryInterceptorTests.java

@@ -21,9 +21,11 @@
 import java.nio.file.AccessDeniedException;
 import java.nio.file.FileSystemException;
 import java.time.Duration;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.assertj.core.api.ThrowingConsumer;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
@@ -316,6 +318,83 @@ void adaptReactiveResultWithAlwaysFailingOperation() {
 	}
 
 
+	@Nested
+	class TimeoutTests {
+
+		private final AnnotatedMethodBean proxy = getProxiedAnnotatedMethodBean();
+		private final AnnotatedMethodBean target = (AnnotatedMethodBean) AopProxyUtils.getSingletonTarget(proxy);
+
+		@Test
+		void timeoutNotExceededAfterInitialSuccess() {
+			String result = proxy.retryOperationWithTimeoutNotExceededAfterInitialSuccess().block();
+			assertThat(result).isEqualTo("success");
+			// 1 initial attempt + 0 retries
+			assertThat(target.counter).hasValue(1);
+		}
+
+		@Test
+		void timeoutNotExceededAndRetriesExhausted() {
+			assertThatIllegalStateException()
+					.isThrownBy(() -> proxy.retryOperationWithTimeoutNotExceededAndRetriesExhausted().block())
+					.satisfies(isRetryExhaustedException())
+					.havingCause()
+						.isInstanceOf(IOException.class)
+						.withMessage("4");
+			// 1 initial attempt + 3 retries
+			assertThat(target.counter).hasValue(4);
+		}
+
+		@Test
+		void timeoutExceededAfterInitialFailure() {
+			assertThatRuntimeException()
+					.isThrownBy(() -> proxy.retryOperationWithTimeoutExceededAfterInitialFailure().block())
+					.satisfies(isReactiveException())
+					.havingCause()
+						.isInstanceOf(TimeoutException.class)
+						.withMessageContaining("within 5ms");
+			// 1 initial attempt + 0 retries
+			assertThat(target.counter).hasValue(1);
+		}
+
+		@Test
+		void timeoutExceededAfterFirstDelayButBeforeFirstRetry() {
+			assertThatRuntimeException()
+					.isThrownBy(() -> proxy.retryOperationWithTimeoutExceededAfterFirstDelayButBeforeFirstRetry().block())
+					.satisfies(isReactiveException())
+					.havingCause()
+						.isInstanceOf(TimeoutException.class)
+						.withMessageContaining("within 5ms");
+			// 1 initial attempt + 0 retries
+			assertThat(target.counter).hasValue(1);
+		}
+
+		@Test
+		void timeoutExceededAfterFirstRetry() {
+			assertThatRuntimeException()
+					.isThrownBy(() -> proxy.retryOperationWithTimeoutExceededAfterFirstRetry().block())
+					.satisfies(isReactiveException())
+					.havingCause()
+						.isInstanceOf(TimeoutException.class)
+						.withMessageContaining("within 5ms");
+			// 1 initial attempt + 1 retry
+			assertThat(target.counter).hasValue(2);
+		}
+
+		@Test
+		void timeoutExceededAfterSecondRetry() {
+			assertThatRuntimeException()
+					.isThrownBy(() -> proxy.retryOperationWithTimeoutExceededAfterSecondRetry().block())
+					.satisfies(isReactiveException())
+					.havingCause()
+						.isInstanceOf(TimeoutException.class)
+						.withMessageContaining("within 5ms");
+			// 1 initial attempt + 2 retries
+			assertThat(target.counter).hasValue(3);
+		}
+
+	}
+
+
 	private static ThrowingConsumer<? super Throwable> isReactiveException() {
 		return ex -> assertThat(ex.getClass().getName()).isEqualTo("reactor.core.Exceptions$ReactiveException");
 	}
@@ -368,6 +447,61 @@ public Mono<Object> retryOperation() {
 				throw new IOException(counter.toString());
 			});
 		}
+
+		@Retryable(timeout = 555, delay = 10)
+		public Mono<String> retryOperationWithTimeoutNotExceededAfterInitialSuccess() {
+			return Mono.fromCallable(() -> {
+				counter.incrementAndGet();
+				return "success";
+			});
+		}
+
+		@Retryable(timeout = 555, delay = 10)
+		public Mono<Object> retryOperationWithTimeoutNotExceededAndRetriesExhausted() {
+			return Mono.fromCallable(() -> {
+				counter.incrementAndGet();
+				throw new IOException(counter.toString());
+			});
+		}
+
+		@Retryable(timeout = 5, delay = 10)
+		public Mono<Object> retryOperationWithTimeoutExceededAfterInitialFailure() {
+			return Mono.fromCallable(() -> {
+				counter.incrementAndGet();
+				Thread.sleep(10);
+				throw new IOException(counter.toString());
+			});
+		}
+
+		@Retryable(timeout = 5, delay = 10)
+		public Mono<Object> retryOperationWithTimeoutExceededAfterFirstDelayButBeforeFirstRetry() {
+			return Mono.fromCallable(() -> {
+				counter.incrementAndGet();
+				throw new IOException(counter.toString());
+			});
+		}
+
+		@Retryable(timeout = 5, delay = 0)
+		public Mono<Object> retryOperationWithTimeoutExceededAfterFirstRetry() {
+			return Mono.fromCallable(() -> {
+				counter.incrementAndGet();
+				if (counter.get() == 2) {
+					Thread.sleep(10);
+				}
+				throw new IOException(counter.toString());
+			});
+		}
+
+		@Retryable(timeout = 5, delay = 0)
+		public Mono<Object> retryOperationWithTimeoutExceededAfterSecondRetry() {
+			return Mono.fromCallable(() -> {
+				counter.incrementAndGet();
+				if (counter.get() == 3) {
+					Thread.sleep(10);
+				}
+				throw new IOException(counter.toString());
+			});
+		}
 	}