Adapt Kafka auto-configuration to core retry semantics

This commit adapts the auto-configuration of retry topics to Spring
Kafka moving away from Spring Retry. The random property has been
removed in favor of the new jitter value and default values have been
made more explicit.

Closes gh-47125

Author: snicoll

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfiguration.java

@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Map;
+import java.util.function.Predicate;
 
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -47,6 +48,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.ImportRuntimeHints;
+import org.springframework.core.retry.RetryPolicy;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -62,8 +64,6 @@
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.util.StringUtils;
 import org.springframework.util.backoff.BackOff;
-import org.springframework.util.backoff.ExponentialBackOff;
-import org.springframework.util.backoff.FixedBackOff;
 
 /**
  * {@link EnableAutoConfiguration Auto-configuration} for Apache Kafka.
@@ -193,13 +193,13 @@ KafkaAdmin kafkaAdmin(KafkaConnectionDetails connectionDetails) {
 	@ConditionalOnSingleCandidate(KafkaTemplate.class)
 	RetryTopicConfiguration kafkaRetryTopicConfiguration(KafkaTemplate<?, ?> kafkaTemplate) {
 		KafkaProperties.Retry.Topic retryTopic = this.properties.getRetry().getTopic();
-		return RetryTopicConfigurationBuilder.newInstance()
+		RetryTopicConfigurationBuilder builder = RetryTopicConfigurationBuilder.newInstance()
 			.maxAttempts(retryTopic.getAttempts())
 			.useSingleTopicForSameIntervals()
 			.suffixTopicsWithIndexValues()
 			.doNotAutoCreateRetryTopics()
-			.customBackoff(getBackOffPolicy(retryTopic.getBackoff()))
-			.create(kafkaTemplate);
+			.customBackoff(getBackOff(retryTopic.getBackoff()));
+		return builder.create(kafkaTemplate);
 	}
 
 	private void applyKafkaConnectionDetailsForConsumer(Map<String, Object> properties,
@@ -226,28 +226,14 @@ private void applyKafkaConnectionDetailsForAdmin(Map<String, Object> properties,
 		applySslBundle(properties, admin.getSslBundle());
 	}
 
-	private BackOff getBackOffPolicy(Backoff properties) {
-		Duration delay = properties.getDelay();
-		Duration maxDelay = properties.getMaxDelay();
-		if (delay == null || Duration.ZERO.equals(delay)) {
-			return new FixedBackOff(0);
-		}
-		if (properties.getMultiplier() > 0 || (maxDelay != null && maxDelay.toMillis() > delay.toMillis())) {
-			long jitter = 0;
-			if (properties.isRandom() && maxDelay != null) {
-				jitter = (maxDelay.toMillis() - delay.toMillis()) / 2;
-			}
-			ExponentialBackOff backOff = new ExponentialBackOff();
-			backOff.setInitialInterval(delay.toMillis() + jitter);
-			backOff.setJitter(jitter);
-			backOff.setMultiplier(properties.getMultiplier());
-			if (maxDelay != null && maxDelay.toMillis() > delay.toMillis()) {
-				backOff.setMaxInterval(properties.getMaxDelay().toMillis());
-			}
-			return backOff;
-		}
-		return new FixedBackOff(delay.toMillis());
-
+	static BackOff getBackOff(Backoff retryTopicBackoff) {
+		PropertyMapper map = PropertyMapper.get();
+		RetryPolicy.Builder builder = RetryPolicy.builder().maxAttempts(Long.MAX_VALUE);
+		map.from(retryTopicBackoff.getDelay()).to(builder::delay);
+		map.from(retryTopicBackoff.getMaxDelay()).when(Predicate.not(Duration::isZero)).to(builder::maxDelay);
+		map.from(retryTopicBackoff.getMultiplier()).to(builder::multiplier);
+		map.from(retryTopicBackoff.getJitter()).when((Predicate.not(Duration::isZero))).to(builder::jitter);
+		return builder.build().getBackOff();
 	}
 
 	static void applySslBundle(Map<String, Object> properties, @Nullable SslBundle sslBundle) {

module/spring-boot-kafka/src/main/java/org/springframework/boot/kafka/autoconfigure/KafkaProperties.java

@@ -1595,26 +1595,31 @@ public Backoff getBackoff() {
 			public static class Backoff {
 
 				/**
-				 * Canonical backoff period. Used as an initial value in the exponential
-				 * case, and as a minimum value in the uniform case.
+				 * Base delay after the initial invocation. Can be combined with a
+				 * "multiplier" to use an exponential back off strategy.
 				 */
 				private Duration delay = Duration.ofSeconds(1);
 
 				/**
-				 * Multiplier to use for generating the next backoff delay.
+				 * Multiplier for a delay for the next retry attempt, applied to the
+				 * previous delay, starting with the initial delay as well as to the
+				 * applicable jitter for each attempt. Fixed delay by default.
 				 */
-				private double multiplier = 0.0;
+				private double multiplier = 1.0;
 
 				/**
-				 * Maximum wait between retries. If less than the delay then the default
-				 * of 30 seconds is applied.
+				 * Maximum delay for any retry attempt, limiting how far jitter and the
+				 * multiplier can increase the delay.
 				 */
-				private Duration maxDelay = Duration.ZERO;
+				private Duration maxDelay = Duration.ofSeconds(30);
 
 				/**
-				 * Whether to have the backoff delays.
+				 * Jitter value for the base retry attempt, randomly subtracted or added
+				 * to the calculated delay, resulting in a value between 'delay - jitter'
+				 * and 'delay + jitter' but never below the base delay or above the max
+				 * delay.
 				 */
-				private boolean random = false;
+				private Duration jitter = Duration.ZERO;
 
 				public Duration getDelay() {
 					return this.delay;
@@ -1640,12 +1645,12 @@ public void setMaxDelay(Duration maxDelay) {
 					this.maxDelay = maxDelay;
 				}
 
-				public boolean isRandom() {
-					return this.random;
+				public Duration getJitter() {
+					return this.jitter;
 				}
 
-				public void setRandom(boolean random) {
-					this.random = random;
+				public void setJitter(Duration jitter) {
+					this.jitter = jitter;
 				}
 
 			}

module/spring-boot-kafka/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -118,6 +118,14 @@
         "level": "error"
       }
     },
+    {
+      "name": "spring.kafka.retry.topic.backoff.random",
+      "type": "java.lang.Boolean",
+      "deprecation": {
+        "replacement": "spring.kafka.retry.topic.backoff.jitter",
+        "level": "error"
+      }
+    },
     {
       "name": "spring.kafka.retry.topic.delay",
       "type": "java.time.Duration",

module/spring-boot-kafka/src/test/java/org/springframework/boot/kafka/autoconfigure/KafkaAutoConfigurationTests.java

@@ -50,6 +50,7 @@
 import org.springframework.aot.hint.predicate.RuntimeHintsPredicates;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.autoconfigure.ssl.SslAutoConfiguration;
+import org.springframework.boot.kafka.autoconfigure.KafkaProperties.Retry;
 import org.springframework.boot.ssl.SslBundle;
 import org.springframework.boot.ssl.SslStoreBundle;
 import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
@@ -96,6 +97,7 @@
 import org.springframework.kafka.transaction.KafkaAwareTransactionManager;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.util.backoff.ExponentialBackOff;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
@@ -568,6 +570,18 @@ void streamsWithCustomKafkaConfiguration() {
 			});
 	}
 
+	@Test
+	void getBackOffWithDefaultsMatchesMapping() {
+		Retry.Topic.Backoff properties = new Retry.Topic.Backoff();
+		assertThat(KafkaAutoConfiguration.getBackOff(properties)).isInstanceOfSatisfying(ExponentialBackOff.class,
+				(backOff) -> {
+					assertThat(backOff.getInitialInterval()).isEqualTo(properties.getDelay().toMillis());
+					assertThat(backOff.getMultiplier()).isEqualTo(properties.getMultiplier());
+					assertThat(backOff.getMaxInterval()).isEqualTo(properties.getMaxDelay().toMillis());
+					assertThat(backOff.getJitter()).isEqualTo(properties.getJitter().toMillis());
+				});
+	}
+
 	@Test
 	void retryTopicConfigurationIsNotEnabledByDefault() {
 		this.contextRunner