GH-4051: Replace Spring Retry usage to core retry

Fixes: #4051

This commit replaces Spring Retry by the core retry support introduced
in Spring Framework 7. This is a breaking change mostly in configuration
that is detailed below.

The main feature in Spring Kafka is BackOffValuesGenerator that
generates the required BackOff values upfront. These are then managed by
the listener infrastructure and Spring Retry is no longer involved.

Moving this code from BackOffPolicy to BackOff dramatically simplifies
that class as Spring Framework's core API naturally provides this
information without the need of an extra infrastructure.

From a configuration standpoint, Spring Kafka relies quite heavily on
Spring Retry's `@Backoff`. As there is no equivalent, the annotation has
been moved to Spring Kafka proper with the following improvements:

* Harmonized name (`@BackOff` instead of `@Backoff`).
* Revisited Javadoc.
* Support for expression evaluation and `java.util.Duration` format.

The creation of a `BackOff` instance from the annotation is now
isolated in `BackOffFactory` and the relevant tests have been added.
`RetryTopicConfigurationBuilder` is mostly backward-compatible but
`uniformRandomBackoff` has been deprecated as we feel that its name
does not convey what it actually does.

`RetryingDeserializer` no longer offer a `RecoveryCallback` but an
equivalent function that takes `RetryException` as an input. This
contains the exceptions thrown as well as the number of retry
attempts.

The use of BinaryExceptionClassifier has been replaced by the newly
introduced `ExceptionMatcher` that is a copy of the original
algorithm with a polished API.

With the migration done, we believe that further improvements can be
made here: `@BackOff` oddly looks like Spring Framework's `@Retryable`.
As a matter of a fact, the `maxAttempts` and `includes`/`excludes` from
`@RetryableTopic` are touching the same concepts. One option would be to
open up `@Retryable` so that it can be used in more case.

Another area of improvement is that harmonization of BackOff as a term.
It is named "Backoff" in several places, including in `@RetryableTopic`,
and it would be nice if the concept had the same syntax everywhere.

With Spring Retry being completely removed, this commit also removes the
dependency and any further references to it.

Harmonize BackOff term in code and documentation

Some code style cleanup

Signed-off-by: Stéphane Nicoll <[email protected]>

Author: snicoll

build.gradle

@@ -71,7 +71,6 @@ ext {
 	scalaVersion = '2.13'
 	springBootVersion = '3.5.0' // docs module
 	springDataVersion = '2025.1.0-SNAPSHOT'
-	springRetryVersion = '2.0.12'
 	springVersion = '7.0.0-SNAPSHOT'
 
 	idPrefix = 'kafka'
@@ -249,9 +248,6 @@ project ('spring-kafka') {
 		api 'org.springframework:spring-context'
 		api 'org.springframework:spring-messaging'
 		api 'org.springframework:spring-tx'
-		api ("org.springframework.retry:spring-retry:$springRetryVersion") {
-			exclude group: 'org.springframework'
-		}
 		api "org.apache.kafka:kafka-clients:$kafkaVersion"
 		api 'io.micrometer:micrometer-observation'
 		optionalApi "org.apache.kafka:kafka-streams:$kafkaVersion"
@@ -322,7 +318,6 @@ project ('spring-kafka-test') {
 		api "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion"
 		api 'org.springframework:spring-context'
 		api 'org.springframework:spring-test'
-		api "org.springframework.retry:spring-retry:$springRetryVersion"
 
 		api "org.apache.kafka:kafka-clients:$kafkaVersion:test"
 		api "org.apache.kafka:kafka-server:$kafkaVersion"

samples/sample-04/src/main/java/com/example/Application.java

@@ -26,7 +26,7 @@
 import org.springframework.kafka.annotation.RetryableTopic;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.messaging.handler.annotation.Header;
-import org.springframework.retry.annotation.Backoff;
+import org.springframework.kafka.annotation.BackOff;
 
 /**
  * Sample shows use of topic-based retry.
@@ -44,7 +44,7 @@ public static void main(String[] args) {
 		SpringApplication.run(Application.class, args);
 	}
 
-	@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 2_000, maxDelay = 10_000, multiplier = 2))
+	@RetryableTopic(attempts = "5", backOff = @BackOff(delay = 2_000, maxDelay = 10_000, multiplier = 2))
 	@KafkaListener(id = "fooGroup", topics = "topic4")
 	public void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
 			@Header(KafkaHeaders.OFFSET) long offset) {

spring-kafka-docs/src/main/antora/modules/ROOT/pages/appendix/change-history.adoc

@@ -236,8 +236,8 @@ See xref:kafka/serdes.adoc#error-handling-deserializer[Using `ErrorHandlingDeser
 
 [[x31-retryable]]
 === Retryable Topics
-Change suffix `-retry-5000` to `-retry` when `@RetryableTopic(backoff = @Backoff(delay = 5000), attempts = "2", fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)`.
-If you want to keep suffix `-retry-5000`, use `@RetryableTopic(backoff = @Backoff(delay = 5000), attempts = "2")`.
+Change suffix `-retry-5000` to `-retry` when `@RetryableTopic(backOff = @BackOff(delay = 5000), attempts = "2", fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)`.
+If you want to keep suffix `-retry-5000`, use `@RetryableTopic(backOff = @BackOff(delay = 5000), attempts = "2")`.
 See xref:retrytopic/topic-naming.adoc[Topic Naming] for more information.
 
 [[x31-c]]

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/annotation-error-handling.adoc

@@ -228,7 +228,7 @@ The exceptions that are considered fatal, by default, are:
 since these exceptions are unlikely to be resolved on a retried delivery.
 
 You can add more exception types to the not-retryable category, or completely replace the map of classified exceptions.
-See the Javadocs for `DefaultErrorHandler.addNotRetryableException()` and `DefaultErrorHandler.setClassifications()` for more information, as well as those for the `spring-retry` `BinaryExceptionClassifier`.
+See the Javadocs for `DefaultErrorHandler.addNotRetryableException()` and `DefaultErrorHandler.setClassifications()` for more information, as well as `ExceptionMatcher`.
 
 Here is an example that adds `IllegalArgumentException` to the not-retryable exceptions:
 
@@ -502,7 +502,7 @@ The exceptions that are considered fatal, by default, are:
 since these exceptions are unlikely to be resolved on a retried delivery.
 
 You can add more exception types to the not-retryable category, or completely replace the map of classified exceptions.
-See the Javadocs for `DefaultAfterRollbackProcessor.setClassifications()` for more information, as well as those for the `spring-retry` `BinaryExceptionClassifier`.
+See the Javadocs for `DefaultAfterRollbackProcessor.setClassifications()` for more information, as well as `ExceptionMatcher`.
 
 Here is an example that adds `IllegalArgumentException` to the not-retryable exceptions:
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/serdes.adoc

@@ -400,9 +400,9 @@ ConsumerFactory cf = new DefaultKafkaConsumerFactory(myConsumerConfigs,
     new RetryingDeserializer(myUnreliableValueDeserializer, retryTemplate));
 ----
 
-Starting with version `3.1.2`, a `RecoveryCallback` can be set on the `RetryingDeserializer` optionally.
+A recovery callback be set on the `RetryingDeserializer`, to return a fallback object if all retries are exhausted.
 
-Refer to the https://github.com/spring-projects/spring-retry[spring-retry] project for configuration of the `RetryTemplate` with a retry policy, back off policy, etc.
+Refer to the https://github.com/spring-projects/spring-framework[Spring Framework] project for configuration of the `RetryTemplate` with a retry policy, back off, etc.
 
 
 [[messaging-message-conversion]]

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/accessing-delivery-attempts.adoc

@@ -17,7 +17,7 @@ Starting with version 3.0.10, a convenient `KafkaMessageHeaderAccessor` is provi
 
 [souce, java]
 ----
-@RetryableTopic(backoff = @Backoff(...))
+@RetryableTopic(backOff = @BackOff(...))
 @KafkaListener(id = "dh1", topics = "dh1")
 void listen(Thing thing, KafkaMessageHeaderAccessor accessor) {
     ...

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/features.adoc

@@ -20,7 +20,7 @@ It includes:
 [source, java]
 ----
 @RetryableTopic(attempts = 5,
-    backoff = @Backoff(delay = 1000, multiplier = 2, maxDelay = 5000))
+    backOff = @BackOff(delay = 1000, multiplier = 2, maxDelay = 5000))
 @KafkaListener(topics = "my-annotated-topic")
 public void processMessage(MyPojo message) {
     // ... message processing
@@ -68,7 +68,7 @@ If that time is reached, the next time the consumer throws an exception the mess
 
 [source, java]
 ----
-@RetryableTopic(backoff = @Backoff(2_000), timeout = 5_000)
+@RetryableTopic(backOff = @BackOff(2_000), timeout = 5_000)
 @KafkaListener(topics = "my-annotated-topic")
 public void processMessage(MyPojo message) {
     // ... message processing

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/topic-naming.adoc

@@ -79,7 +79,7 @@ NOTE: The previous `FixedDelayStrategy` is now deprecated, and can be replaced b
 
 [source, java]
 ----
-@RetryableTopic(backoff = @Backoff(2_000), sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+@RetryableTopic(backOff = @BackOff(2_000), sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
 @KafkaListener(topics = "my-annotated-topic")
 public void processMessage(MyPojo message) {
     // ... message processing
@@ -138,7 +138,7 @@ If multiple topics are required, then that can be done using the following confi
 [source, java]
 ----
 @RetryableTopic(attempts = 230,
-    backoff = @Backoff(delay = 1_000, multiplier = 2, maxDelay = 16_000),
+    backOff = @BackOff(delay = 1_000, multiplier = 2, maxDelay = 16_000),
     sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
 @KafkaListener(topics = "my-annotated-topic")
 public void processMessage(MyPojo message) {

spring-kafka/src/main/java/org/springframework/kafka/annotation/BackOff.java

@@ -0,0 +1,200 @@
+/*
+ * Copyright 2018-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.annotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.core.annotation.AliasFor;
+import org.springframework.core.retry.RetryPolicy;
+import org.springframework.format.annotation.DurationFormat;
+import org.springframework.util.backoff.ExponentialBackOff;
+import org.springframework.util.backoff.FixedBackOff;
+
+/**
+ * Collects metadata for creating a {@link org.springframework.util.backoff.BackOff BackOff}
+ * instance as part of a {@link RetryPolicy}. Values can be provided as is or using a
+ * {@code *String} equivalent that supports more format, as well as expression evaluations.
+ * <p>
+ * The available attributes lead to the following:
+ * <ul>
+ * <li>With no explicit settings, the default is a {@link FixedBackOff} with a delay of
+ * {@value #DEFAULT_DELAY} ms</li>
+ * <li>With only {@link #delay()} set: a fixed delay back off with that value</li>
+ * <li>In all other cases, an {@link ExponentialBackOff} is created with the values of
+ * {@link #delay()} (default: {@value RetryPolicy.Builder#DEFAULT_DELAY} ms),
+ * {@link #maxDelay()} (default: no maximum), {@link #multiplier()}
+ * (default: {@value RetryPolicy.Builder#DEFAULT_MULTIPLIER}) and {@link #jitter()}
+ * (default: no jitter).</li>
+ * </ul>
+ *
+ * @author Dave Syer
+ * @author Gary Russell
+ * @author Aftab Shaikh
+ * @author Stephane Nicoll
+ *
+ * @since 4.0
+ */
+@Target(ElementType.ANNOTATION_TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+public @interface BackOff {
+
+	/**
+	 * Default {@link #delay()} in milliseconds.
+	 */
+	long DEFAULT_DELAY = 1000;
+
+	/**
+	 * Alias for {@link #delay()}.
+	 * <p>Intended to be used when no other attributes are needed, for example:
+	 * {@code @BackOff(2000)}.
+	 *
+	 * @return the based delay in milliseconds (default{@value DEFAULT_DELAY})
+	 */
+	@AliasFor("delay")
+	long value() default DEFAULT_DELAY;
+
+	/**
+	 * Specify the base delay after the initial invocation.
+	 * <p>If only a {@code delay} is specified, a {@link FixedBackOff} with that value
+	 * as the interval is configured.
+	 * <p>If a {@linkplain #multiplier() multiplier} is specified, this serves as the
+	 * initial delay to multiply from.
+	 * <p>The default is {@value DEFAULT_DELAY} milliseconds.
+	 *
+	 * @return the based delay in milliseconds (default{@value DEFAULT_DELAY})
+	 */
+	@AliasFor("value")
+	long delay() default DEFAULT_DELAY;
+
+	/**
+	 * Specify the base delay after the initial invocation using a String format. If
+	 * this is specified, takes precedence over {@link #delay()}.
+	 * <p>The delay String can be in several formats:
+	 * <ul>
+	 * <li>a plain long &mdash; which is interpreted to represent a duration in
+	 * milliseconds</li>
+	 * <li>any of the known {@link DurationFormat.Style}: the {@link DurationFormat.Style#ISO8601 ISO8601}
+	 * style or the {@link DurationFormat.Style#SIMPLE SIMPLE} style &mdash; using
+	 * milliseconds as fallback if the string doesn't contain an explicit unit</li>
+	 * <li>Regular expressions, such as {@code ${example.property}} to use the
+	 * {@code example.property} from the environment</li>
+	 * </ul>
+	 *
+	 * @return the based delay as a String value &mdash; for example a placeholder
+	 * @see #delay()
+	 */
+	String delayString() default "";
+
+	/**
+	 * Specify the maximum delay for any retry attempt, limiting how far
+	 * {@linkplain #jitter jitter} and the {@linkplain #multiplier() multiplier} can
+	 * increase the {@linkplain #delay() delay}.
+	 * <p>Ignored if only {@link #delay()} is set, otherwise an {@link ExponentialBackOff}
+	 * with the given max delay or an unlimited delay if not set.
+	 *
+	 * @return the maximum delay
+	 */
+	long maxDelay() default 0;
+
+	/**
+	 * Specify the maximum delay for any retry attempt using a String format. If this is
+	 * specified, takes precedence over {@link #maxDelay()}..
+	 * <p>The max delay String can be in several formats:
+	 * <ul>
+	 * <li>a plain long &mdash; which is interpreted to represent a duration in
+	 * milliseconds</li>
+	 * <li>any of the known {@link DurationFormat.Style}: the {@link DurationFormat.Style#ISO8601 ISO8601}
+	 * style or the {@link DurationFormat.Style#SIMPLE SIMPLE} style &mdash; using
+	 * milliseconds as fallback if the string doesn't contain an explicit unit</li>
+	 * <li>Regular expressions, such as {@code ${example.property}} to use the
+	 * {@code example.property} from the environment</li>
+	 * </ul>
+	 *
+	 * @return the max delay as a String value &mdash; for example a placeholder
+	 * @see #maxDelay()
+	 */
+	String maxDelayString() default "";
+
+	/**
+	 * Specify a multiplier for a delay for the next retry attempt, applied to the previous
+	 * delay, starting with the initial {@linkplain #delay() delay} as well as to the
+	 * applicable {@linkplain #jitter() jitter} for each attempt.
+	 * <p>Ignored if only {@link #delay()} is set, otherwise an {@link ExponentialBackOff}
+	 * with the given multiplier or {@code 1.0} if not set.
+	 *
+	 * @return the value to multiply the current interval by for each attempt
+	 */
+	double multiplier() default 0;
+
+	/**
+	 * Specify a multiplier for a delay for the next retry attempt using a String format.
+	 * If this is specified, takes precedence over {@link #multiplier()}.
+	 * <p>The multiplier String can be in several formats:
+	 * <ul>
+	 * <li>a plain double</li>
+	 * <li>Regular expressions, such as {@code ${example.property}} to use the
+	 * {@code example.property} from the environment</li>
+	 * </ul>
+	 *
+	 * @return the value to multiply the current interval by for each attempt &mdash;
+	 * for example, a placeholder
+	 * @see #multiplier()
+	 */
+	String multiplierString() default "";
+
+	/**
+	 * Specify a jitter value for the base retry attempt, randomly subtracted or added to
+	 * the calculated delay, resulting in a value between {@code delay - jitter} and
+	 * {@code delay + jitter} but never below the {@linkplain #delay() base delay} or
+	 * above the {@linkplain #maxDelay() max delay}.
+	 * <p>If a {@linkplain #multiplier() multiplier} is specified, it is applied to the
+	 * jitter value as well.
+	 * <p>Ignored if only {@link #delay()} is set, otherwise an {@link ExponentialBackOff}
+	 * with the given jitter or no jitter if not set.
+	 *
+	 * @return the jitter value in milliseconds
+	 * @see #delay()
+	 * @see #maxDelay()
+	 * @see #multiplier()
+	 */
+	long jitter() default 0;
+
+	/**
+	 * Specify a jitter value for the base retry attempt using a String format. If this is
+	 * specified, takes precedence over {@link #jitter()}.
+	 * <p>The jitter String can be in several formats:
+	 * <ul>
+	 * <li>a plain long &mdash; which is interpreted to represent a duration in
+	 * milliseconds</li>
+	 * <li>any of the known {@link DurationFormat.Style}: the {@link DurationFormat.Style#ISO8601 ISO8601}
+	 * style or the {@link DurationFormat.Style#SIMPLE SIMPLE} style &mdash; using
+	 * milliseconds as fallback if the string doesn't contain an explicit unit</li>
+	 * <li>Regular expressions, such as {@code ${example.property}} to use the
+	 * {@code example.property} from the environment</li>
+	 * </ul>
+	 *
+	 * @return the jitter as a String value &mdash; for example, a placeholder
+	 * @see #jitter()
+	 */
+	String jitterString() default "";
+
+}