Add validation tests

Author: tomazfernandes

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

@@ -169,10 +169,10 @@ private static void setBackOffPolicy(RetryTopicConfigurationBuilder builder, Top
 	}
 
 	private static void validateRetryTopicInput(Topic retryTopic) {
-		assertProperty("attempts",  retryTopic.getAttempts(), 1);
-		assertProperty("delay",  retryTopic.getDelayMillis(), 0);
-		assertProperty("multiplier",  retryTopic.getMultiplier(), 0);
-		assertProperty("maxDelayMillis",  retryTopic.getMaxDelayMillis(), 0);
+		assertProperty("attempts", retryTopic.getAttempts(), 1);
+		assertProperty("delay", retryTopic.getDelayMillis(), 0);
+		assertProperty("multiplier", retryTopic.getMultiplier(), 0);
+		assertProperty("maxDelayMillis", retryTopic.getMaxDelayMillis(), 0);
 	}
 
 	private static void assertProperty(String propertyName, Number providedValue, int minValue) {

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

@@ -369,6 +369,51 @@ void retryTopicConfigurationWithNoBackOff() {
 								.extracting(DestinationTopic.Properties::delay).containsExactly(0L, 0L, 0L));
 	}
 
+	@Test
+	void retryTopicConfigurationWithNegativeDelay() {
+		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=-1")
+				.run((context) -> assertThat(context.getStartupFailure()).getRootCause()
+						.isInstanceOf(IllegalArgumentException.class).message()
+						.isEqualTo("Property spring.kafka.retry.topic.delay"
+								+ " should be greater than or equal to 0. Provided value was -1."));
+	}
+
+	@Test
+	void retryTopicConfigurationWithNegativeMultiplier() {
+		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.multiplier=-1")
+				.run((context) -> assertThat(context.getStartupFailure()).getRootCause()
+						.isInstanceOf(IllegalArgumentException.class).message()
+						.isEqualTo("Property spring.kafka.retry.topic.multiplier"
+								+ " should be greater than or equal to 0. Provided value was -1.0."));
+	}
+
+	@Test
+	void retryTopicConfigurationWithNegativeMaxDelay() {
+		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.maxDelay=-1")
+				.run((context) -> assertThat(context.getStartupFailure()).getRootCause()
+						.isInstanceOf(IllegalArgumentException.class).message()
+						.isEqualTo("Property spring.kafka.retry.topic.maxDelay"
+								+ " should be greater than or equal to 0. Provided value was -1."));
+	}
+
+	@Test
+	void retryTopicConfigurationWithZeroAttempts() {
+		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=0")
+				.run((context) -> assertThat(context.getStartupFailure()).getRootCause()
+						.isInstanceOf(IllegalArgumentException.class).message()
+						.isEqualTo("Property spring.kafka.retry.topic.attempts"
+								+ " should be greater than or equal to 1. Provided value was 0."));
+	}
+
 	@SuppressWarnings("unchecked")
 	@Test
 	void streamsWithSeveralStreamsBuilderFactoryBeans() {