Address review suggestions

Add properties validation
Add noBackOffPolicy for zero delay
Improve javadocs

Author: tomazfernandes

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

@@ -44,7 +44,6 @@
 import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
-import org.springframework.retry.backoff.BackOffPolicy;
 import org.springframework.retry.backoff.BackOffPolicyBuilder;
 import org.springframework.retry.backoff.SleepingBackOffPolicy;
 import org.springframework.util.Assert;
@@ -65,6 +64,8 @@
 @Import({ KafkaAnnotationDrivenConfiguration.class, KafkaStreamsAnnotationDrivenConfiguration.class })
 public class KafkaAutoConfiguration {
 
+	private static final String RETRY_TOPIC_VALIDATION_ERROR_MSG = "Property spring.kafka.retry.topic.%s should be greater than or equal to %s. Provided value was %s.";
+
 	private final KafkaProperties properties;
 
 	public KafkaAutoConfiguration(KafkaProperties properties) {
@@ -150,18 +151,32 @@ public KafkaAdmin kafkaAdmin() {
 	@ConditionalOnProperty(name = "spring.kafka.retry.topic.enabled")
 	public RetryTopicConfiguration kafkaRetryTopicConfiguration(KafkaOperations<Object, Object> kafkaOperations) {
 		KafkaProperties.Retry.Topic retryTopic = this.properties.getRetry().getTopic();
-		return RetryTopicConfigurationBuilder.newInstance().maxAttempts(retryTopic.getAttempts())
-				.customBackoff(getBackOffPolicy(retryTopic)).useSingleTopicForFixedDelays()
-				.suffixTopicsWithIndexValues().doNotAutoCreateRetryTopics().create(kafkaOperations);
+		validateRetryTopicInput(retryTopic);
+		RetryTopicConfigurationBuilder builder = RetryTopicConfigurationBuilder.newInstance()
+				.maxAttempts(retryTopic.getAttempts()).useSingleTopicForFixedDelays().suffixTopicsWithIndexValues()
+				.doNotAutoCreateRetryTopics();
+		setBackOffPolicy(builder, retryTopic);
+		return builder.create(kafkaOperations);
+	}
+
+	private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Topic retryTopic) {
+		PropertyMapper.get().from(retryTopic.getDelayMillis()).whenEqualTo(0L).toCall(builder::noBackoff);
+		PropertyMapper.get().from(retryTopic.getDelayMillis()).when((delay) -> delay > 0)
+				.toCall(() -> builder.customBackoff((SleepingBackOffPolicy<?>) BackOffPolicyBuilder.newBuilder()
+						.delay(retryTopic.getDelayMillis()).maxDelay(retryTopic.getMaxDelayMillis())
+						.multiplier(retryTopic.getMultiplier()).random(retryTopic.isRandomBackOff()).build()));
+	}
+
+	private static void validateRetryTopicInput(Topic retryTopic) {
+		assertProperty("attempts", retryTopic.getAttempts() >= 1, 1, retryTopic.getMaxDelayMillis());
+		assertProperty("delay", retryTopic.getDelayMillis() >= 0, 0, retryTopic.getDelayMillis());
+		assertProperty("multiplier", retryTopic.getMultiplier() >= 0, 0, retryTopic.getMultiplier());
+		assertProperty("maxDelayMillis", retryTopic.getDelayMillis() >= 0, 0, retryTopic.getMaxDelayMillis());
 	}
 
-	private SleepingBackOffPolicy<?> getBackOffPolicy(Topic retryTopic) {
-		BackOffPolicy policy = BackOffPolicyBuilder.newBuilder().delay(retryTopic.getDelayMillis())
-				.maxDelay(retryTopic.getMaxDelayMillis()).multiplier(retryTopic.getMultiplier())
-				.random(retryTopic.isRandomBackOff()).build();
-		Assert.isInstanceOf(SleepingBackOffPolicy.class, policy,
-				() -> "BackOffPolicy must be an instance of SleepingBackOffPolicy. Provided: " + policy);
-		return (SleepingBackOffPolicy<?>) policy;
+	private static void assertProperty(String propertyName, boolean condition, Object minValue, Object providedValue) {
+		Assert.isTrue(condition,
+				() -> String.format(RETRY_TOPIC_VALIDATION_ERROR_MSG, propertyName, minValue, providedValue));
 	}
 
 }

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

@@ -1380,15 +1380,14 @@ public static class Topic {
 			private Double multiplier = 0.0;
 
 			/**
-			 * The maximimum wait between retries. If less than the delay then the default
+			 * The maximum wait between retries. If less than the delay then the default
 			 * of 30 seconds is applied.
 			 */
 			private Duration maxDelay = Duration.ZERO;
 
 			/**
-			 * In the exponential case (multiplier() > 0) set this to true to have the
-			 * backoff delays randomized, so that the maximum delay is multiplier times
-			 * the previous delay and the distribution is uniform between the two values.
+			 * In the exponential case, set this to true to have the backoff delays
+			 * randomized. This has no effect for other back off types.
 			 */
 			private Boolean randomBackOff = false;
 

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

@@ -359,6 +359,16 @@ void retryTopicConfigurationWithFixedBackOff() {
 								.extracting(DestinationTopic.Properties::delay).containsExactly(0L, 2000L, 0L));
 	}
 
+	@Test
+	void retryTopicConfigurationWithNoBackOff() {
+		this.contextRunner.withPropertyValues("spring.application.name=my-test-app",
+				"spring.kafka.bootstrap-servers=localhost:9092,localhost:9093", "spring.kafka.retry.topic.enabled=true",
+				"spring.kafka.retry.topic.attempts=4", "spring.kafka.retry.topic.delay=0")
+				.run((context) -> assertThat(
+						context.getBean(RetryTopicConfiguration.class).getDestinationTopicProperties()).hasSize(3)
+								.extracting(DestinationTopic.Properties::delay).containsExactly(0L, 0L, 0L));
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	void streamsWithSeveralStreamsBuilderFactoryBeans() {