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

spring-core/src/test/java/org/springframework/core/retry/RetryPolicyTests.java

@@ -131,6 +131,22 @@ class RetryPolicyTests {
 			assertToString(policy, 1000, 0, 1.0, Long.MAX_VALUE, 5);
 		}
 
+		@Test
+		void timeoutPreconditions() {
+			assertThatIllegalArgumentException()
+					.isThrownBy(() -> RetryPolicy.builder().timeout(Duration.ofMillis(-1)))
+					.withMessage("Invalid timeout (-1ms): must be greater than or equal to zero.");
+		}
+
+		@Test
+		void timeout() {
+			Duration timeout = Duration.ofMillis(42);
+
+			var policy = RetryPolicy.builder().timeout(timeout).build();
+
+			assertThat(policy.getTimeout()).isSameAs(timeout);
+		}
+
 		@Test
 		void delayPreconditions() {
 			assertThatIllegalArgumentException()

spring-core/src/test/java/org/springframework/core/retry/RetryTemplateTests.java

@@ -26,6 +26,7 @@ import java.util.function.Consumer;
 
 import org.assertj.core.api.ThrowingConsumer;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments.ArgumentSet;
@@ -407,6 +408,185 @@ class RetryTemplateTests {
 	}
 
 
+	@Nested
+	class TimeoutTests {
+
+		@Test
+		void retryWithImmediateSuccessAndTimeoutExceeded() throws Exception {
+			RetryPolicy retryPolicy = RetryPolicy.builder().timeout(Duration.ofMillis(5)).build();
+			RetryTemplate retryTemplate = new RetryTemplate(retryPolicy);
+			retryTemplate.setRetryListener(retryListener);
+
+			AtomicInteger invocationCount = new AtomicInteger();
+			Retryable<String> retryable = () -> {
+				invocationCount.incrementAndGet();
+				Thread.sleep(10);
+				return "always succeeds";
+			};
+
+			assertThat(invocationCount).hasValue(0);
+			assertThat(retryTemplate.execute(retryable)).isEqualTo("always succeeds");
+			assertThat(invocationCount).hasValue(1);
+
+			// RetryListener interactions:
+			verifyNoInteractions(retryListener);
+		}
+
+		@Test
+		void retryWithInitialFailureAndZeroRetriesRetryPolicyAndTimeoutExceeded() {
+			RetryPolicy retryPolicy = RetryPolicy.builder()
+					.timeout(Duration.ofMillis(5))
+					.predicate(throwable -> false) // Zero retries
+					.build();
+			RetryTemplate retryTemplate = new RetryTemplate(retryPolicy);
+			retryTemplate.setRetryListener(retryListener);
+
+			Exception exception = new RuntimeException("Boom!");
+			Retryable<String> retryable = () -> {
+				Thread.sleep(10);
+				throw exception;
+			};
+
+			assertThatExceptionOfType(RetryException.class)
+					.isThrownBy(() -> retryTemplate.execute(retryable))
+					.withMessageMatching("Retry policy for operation '.+?' exhausted; aborting execution")
+					.withCause(exception)
+					.satisfies(throwable -> assertThat(throwable.getSuppressed()).isEmpty())
+					.satisfies(throwable -> assertThat(throwable.getRetryCount()).isZero())
+					.satisfies(throwable -> inOrder.verify(retryListener).onRetryPolicyExhaustion(retryPolicy, retryable, throwable));
+
+			verifyNoMoreInteractions(retryListener);
+		}
+
+		@Test
+		void retryWithTimeoutExceededAfterInitialFailure() throws Exception {
+			RetryPolicy retryPolicy = RetryPolicy.builder()
+					.timeout(Duration.ofMillis(5))
+					.delay(Duration.ZERO)
+					.build();
+			RetryTemplate retryTemplate = new RetryTemplate(retryPolicy);
+			retryTemplate.setRetryListener(retryListener);
+
+			AtomicInteger invocationCount = new AtomicInteger();
+			Retryable<String> retryable = () -> {
+				Thread.sleep(10);
+				throw new CustomException("Boom " + invocationCount.incrementAndGet());
+			};
+
+			assertThat(invocationCount).hasValue(0);
+			assertThatExceptionOfType(RetryException.class)
+					.isThrownBy(() -> retryTemplate.execute(retryable))
+					.withMessageMatching("Retry policy for operation '.+?' exceeded timeout \\(5 ms\\); aborting execution")
+					.withCause(new CustomException("Boom 1"))
+					.satisfies(throwable -> inOrder.verify(retryListener).onRetryPolicyTimeout(
+							eq(retryPolicy), eq(retryable), eq(throwable)));
+			assertThat(invocationCount).hasValue(1);
+
+			verifyNoMoreInteractions(retryListener);
+		}
+
+		@Test
+		void retryWithTimeoutExceededAfterFirstDelayButBeforeFirstRetry() throws Exception {
+			RetryPolicy retryPolicy = RetryPolicy.builder()
+					.timeout(Duration.ofMillis(5))
+					.delay(Duration.ofMillis(10)) // Delay > Timeout
+					.build();
+			RetryTemplate retryTemplate = new RetryTemplate(retryPolicy);
+			retryTemplate.setRetryListener(retryListener);
+
+			AtomicInteger invocationCount = new AtomicInteger();
+			Retryable<String> retryable = () -> {
+				throw new CustomException("Boom " + invocationCount.incrementAndGet());
+			};
+
+			assertThat(invocationCount).hasValue(0);
+			assertThatExceptionOfType(RetryException.class)
+					.isThrownBy(() -> retryTemplate.execute(retryable))
+					.withMessageMatching("Retry policy for operation '.+?' exceeded timeout \\(5 ms\\); aborting execution")
+					.withCause(new CustomException("Boom 1"))
+					.satisfies(throwable -> inOrder.verify(retryListener).onRetryPolicyTimeout(
+							eq(retryPolicy), eq(retryable), eq(throwable)));
+			assertThat(invocationCount).hasValue(1);
+
+			verifyNoMoreInteractions(retryListener);
+		}
+
+		@Test
+		void retryWithTimeoutExceededAfterFirstRetry() throws Exception {
+			RetryPolicy retryPolicy = RetryPolicy.builder()
+					.timeout(Duration.ofMillis(5))
+					.delay(Duration.ZERO)
+					.build();
+			RetryTemplate retryTemplate = new RetryTemplate(retryPolicy);
+			retryTemplate.setRetryListener(retryListener);
+
+			AtomicInteger invocationCount = new AtomicInteger();
+			Retryable<String> retryable = () -> {
+				int currentInvocation = invocationCount.incrementAndGet();
+				if (currentInvocation == 2) {
+					Thread.sleep(10);
+				}
+				throw new CustomException("Boom " + currentInvocation);
+			};
+
+			assertThat(invocationCount).hasValue(0);
+			assertThatExceptionOfType(RetryException.class)
+					.isThrownBy(() -> retryTemplate.execute(retryable))
+					.withMessageMatching("Retry policy for operation '.+?' exceeded timeout \\(5 ms\\); aborting execution")
+					.withCause(new CustomException("Boom 2"))
+					.satisfies(throwable -> {
+						inOrder.verify(retryListener).beforeRetry(retryPolicy, retryable);
+						inOrder.verify(retryListener).onRetryFailure(retryPolicy, retryable, new CustomException("Boom 2"));
+
+						inOrder.verify(retryListener).onRetryPolicyTimeout(
+								eq(retryPolicy), eq(retryable), eq(throwable));
+					});
+			assertThat(invocationCount).hasValue(2);
+
+			verifyNoMoreInteractions(retryListener);
+		}
+
+		@Test
+		void retryWithTimeoutExceededAfterSecondRetry() throws Exception {
+			RetryPolicy retryPolicy = RetryPolicy.builder()
+					.timeout(Duration.ofMillis(5))
+					.delay(Duration.ZERO)
+					.build();
+			RetryTemplate retryTemplate = new RetryTemplate(retryPolicy);
+			retryTemplate.setRetryListener(retryListener);
+
+			AtomicInteger invocationCount = new AtomicInteger();
+			Retryable<String> retryable = () -> {
+				int currentInvocation = invocationCount.incrementAndGet();
+				if (currentInvocation == 3) {
+					Thread.sleep(10);
+				}
+				throw new CustomException("Boom " + currentInvocation);
+			};
+
+			assertThat(invocationCount).hasValue(0);
+			assertThatExceptionOfType(RetryException.class)
+					.isThrownBy(() -> retryTemplate.execute(retryable))
+					.withMessageMatching("Retry policy for operation '.+?' exceeded timeout \\(5 ms\\); aborting execution")
+					.withCause(new CustomException("Boom 3"))
+					.satisfies(throwable -> {
+						var counter = new AtomicInteger(1);
+						repeat(2, () -> {
+							inOrder.verify(retryListener).beforeRetry(retryPolicy, retryable);
+							inOrder.verify(retryListener).onRetryFailure(retryPolicy, retryable,
+									new CustomException("Boom " + counter.incrementAndGet()));
+						});
+						inOrder.verify(retryListener).onRetryPolicyTimeout(
+								eq(retryPolicy), eq(retryable), eq(throwable));
+					});
+			assertThat(invocationCount).hasValue(3);
+
+			verifyNoMoreInteractions(retryListener);
+		}
+
+	}
+
+
 	private static void repeat(int times, Runnable runnable) {
 		for (int i = 0; i < times; i++) {
 			runnable.run();

spring-core/src/test/java/org/springframework/core/retry/support/CompositeRetryListenerTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.core.retry.support;
 
+import java.time.Duration;
 import java.util.List;
 
 import org.junit.jupiter.api.BeforeEach;
@@ -102,4 +103,15 @@ class CompositeRetryListenerTests {
 		verify(listener3).onRetryPolicyInterruption(retryPolicy, retryable, exception);
 	}
 
+	@Test
+	void onRetryPolicyTimeout() {
+		Duration elapsedTime = Duration.ofMillis(100);
+		RetryException exception = new RetryException("", new Exception());
+		compositeRetryListener.onRetryPolicyTimeout(retryPolicy, retryable, exception);
+
+		verify(listener1).onRetryPolicyTimeout(retryPolicy, retryable, exception);
+		verify(listener2).onRetryPolicyTimeout(retryPolicy, retryable, exception);
+		verify(listener3).onRetryPolicyTimeout(retryPolicy, retryable, exception);
+	}
+
 }