Polish "Add auto-configuration to Kafka Retry Topics"

See gh-29812

Author: snicoll

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

@@ -17,13 +17,15 @@
 package org.springframework.boot.autoconfigure.kafka;
 
 import java.io.IOException;
+import java.time.Duration;
 
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnSingleCandidate;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Jaas;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties.Retry.Topic;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -34,7 +36,6 @@
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaAdmin;
-import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.retrytopic.RetryTopicConfiguration;
@@ -146,21 +147,30 @@ public KafkaAdmin kafkaAdmin() {
 
 	@Bean
 	@ConditionalOnProperty(name = "spring.kafka.retry.topic.enabled")
-	public RetryTopicConfiguration kafkaRetryTopicConfiguration(KafkaOperations<Object, Object> kafkaOperations) {
+	@ConditionalOnSingleCandidate(KafkaTemplate.class)
+	public RetryTopicConfiguration kafkaRetryTopicConfiguration(KafkaTemplate<?, ?> kafkaTemplate) {
 		KafkaProperties.Retry.Topic retryTopic = this.properties.getRetry().getTopic();
 		RetryTopicConfigurationBuilder builder = RetryTopicConfigurationBuilder.newInstance()
 				.maxAttempts(retryTopic.getAttempts()).useSingleTopicForFixedDelays().suffixTopicsWithIndexValues()
 				.doNotAutoCreateRetryTopics();
 		setBackOffPolicy(builder, retryTopic);
-		return builder.create(kafkaOperations);
+		return builder.create(kafkaTemplate);
 	}
 
 	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()));
+		long delay = (retryTopic.getDelay() != null) ? retryTopic.getDelay().toMillis() : 0;
+		if (delay > 0) {
+			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+			BackOffPolicyBuilder backOffPolicy = BackOffPolicyBuilder.newBuilder();
+			map.from(delay).to(backOffPolicy::delay);
+			map.from(retryTopic.getMaxDelay()).as(Duration::toMillis).to(backOffPolicy::maxDelay);
+			map.from(retryTopic.getMultiplier()).to(backOffPolicy::multiplier);
+			map.from(retryTopic.isRandomBackOff()).to(backOffPolicy::random);
+			builder.customBackoff((SleepingBackOffPolicy<?>) backOffPolicy.build());
+		}
+		else {
+			builder.noBackoff();
+		}
 	}
 
 }

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

@@ -17,7 +17,6 @@
 package org.springframework.boot.autoconfigure.kafka;
 
 import java.io.IOException;
-import java.math.BigDecimal;
 import java.time.Duration;
 import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
@@ -42,7 +41,6 @@
 import org.springframework.core.io.Resource;
 import org.springframework.kafka.listener.ContainerProperties.AckMode;
 import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
-import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.unit.DataSize;
 
@@ -1341,96 +1339,105 @@ public void setOptions(Map<String, String> options) {
 
 	}
 
-	public static class Retry {
+	public static class Security {
 
-		private Topic topic = new Topic();
+		/**
+		 * Security protocol used to communicate with brokers.
+		 */
+		private String protocol;
 
-		public Topic getTopic() {
-			return this.topic.validate();
+		public String getProtocol() {
+			return this.protocol;
+		}
+
+		public void setProtocol(String protocol) {
+			this.protocol = protocol;
+		}
+
+		public Map<String, Object> buildProperties() {
+			Properties properties = new Properties();
+			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
+			map.from(this::getProtocol).to(properties.in(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
+			return properties;
 		}
 
-		public void setTopic(Topic topic) {
-			this.topic = topic;
+	}
+
+	public static class Retry {
+
+		private final Topic topic = new Topic();
+
+		public Topic getTopic() {
+			return this.topic;
 		}
 
 		/**
 		 * Properties for non-blocking, topic-based retries.
 		 */
 		public static class Topic {
 
-			private static final String RETRY_TOPIC_PROPERTIES_PREFIX = "spring.kafka.retry.topic.";
-
-			private static final String RETRY_TOPIC_VALIDATION_ERROR_MSG = "Property " + RETRY_TOPIC_PROPERTIES_PREFIX
-					+ "%s should be greater than or equal to %s. Provided value was %s.";
-
 			/**
-			 * Whether to enable topic-based retries auto-configuration.
+			 * Whether to enable topic-based non-blocking retries.
 			 */
-			private Boolean enabled;
+			private boolean enabled;
 
 			/**
-			 * The total number of processing attempts made before sending the message to
-			 * the DLT.
+			 * Total number of processing attempts made before sending the message to the
+			 * DLT.
 			 */
-			private Integer attempts = 3;
+			private int attempts = 3;
 
 			/**
-			 * A canonical backoff period. Used as an initial value in the exponential
-			 * case, and as a minimum value in the uniform case.
+			 * Canonical backoff period. Used as an initial value in the exponential case,
+			 * and as a minimum value in the uniform case.
 			 */
 			private Duration delay = Duration.ofSeconds(1);
 
 			/**
-			 * If positive, then used as a multiplier for generating the next delay for
-			 * backoff.
+			 * Multiplier to use for generating the next backoff delay.
 			 */
-			private Double multiplier = 0.0;
+			private double multiplier = 0.0;
 
 			/**
-			 * The maximum wait between retries. If less than the delay then the default
-			 * of 30 seconds is applied.
+			 * 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, set this to true to have the backoff delays
-			 * randomized.
+			 * Whether to have the backoff delays.
 			 */
-			private Boolean randomBackOff = false;
+			private boolean randomBackOff = false;
 
-			public Boolean getEnabled() {
+			public boolean isEnabled() {
 				return this.enabled;
 			}
 
-			public void setEnabled(Boolean enabled) {
+			public void setEnabled(boolean enabled) {
 				this.enabled = enabled;
 			}
 
-			public Integer getAttempts() {
+			public int getAttempts() {
 				return this.attempts;
 			}
 
-			public void setAttempts(Integer attempts) {
+			public void setAttempts(int attempts) {
 				this.attempts = attempts;
 			}
 
 			public Duration getDelay() {
 				return this.delay;
 			}
 
-			public Long getDelayMillis() {
-				return (this.delay != null) ? this.delay.toMillis() : null;
-			}
-
 			public void setDelay(Duration delay) {
 				this.delay = delay;
 			}
 
-			public Double getMultiplier() {
+			public double getMultiplier() {
 				return this.multiplier;
 			}
 
-			public void setMultiplier(Double multiplier) {
+			public void setMultiplier(double multiplier) {
 				this.multiplier = multiplier;
 			}
 
@@ -1442,59 +1449,14 @@ public void setMaxDelay(Duration maxDelay) {
 				this.maxDelay = maxDelay;
 			}
 
-			public Long getMaxDelayMillis() {
-				return (this.maxDelay != null) ? this.maxDelay.toMillis() : null;
-			}
-
-			public Boolean isRandomBackOff() {
+			public boolean isRandomBackOff() {
 				return this.randomBackOff;
 			}
 
-			public void setRandomBackOff(Boolean randomBackOff) {
+			public void setRandomBackOff(boolean randomBackOff) {
 				this.randomBackOff = randomBackOff;
 			}
 
-			private Topic validate() {
-				validateProperty("attempts", this.attempts, 1);
-				validateProperty("delay", this.getDelayMillis(), 0);
-				validateProperty("multiplier", this.multiplier, 0);
-				validateProperty("maxDelay", this.getMaxDelayMillis(), 0);
-				Assert.isTrue(this.multiplier != 0 || !this.isRandomBackOff(),
-						"Property " + RETRY_TOPIC_PROPERTIES_PREFIX
-								+ "randomBackOff should not be true with non-exponential back offs.");
-				return this;
-			}
-
-			private static void validateProperty(String propertyName, Number providedValue, int minValue) {
-				Assert.notNull(providedValue, () -> RETRY_TOPIC_PROPERTIES_PREFIX + propertyName + " cannot be null.");
-				Assert.isTrue(new BigDecimal(providedValue.toString()).compareTo(BigDecimal.valueOf(minValue)) >= 0,
-						() -> String.format(RETRY_TOPIC_VALIDATION_ERROR_MSG, propertyName, minValue, providedValue));
-			}
-
-		}
-
-	}
-
-	public static class Security {
-
-		/**
-		 * Security protocol used to communicate with brokers.
-		 */
-		private String protocol;
-
-		public String getProtocol() {
-			return this.protocol;
-		}
-
-		public void setProtocol(String protocol) {
-			this.protocol = protocol;
-		}
-
-		public Map<String, Object> buildProperties() {
-			Properties properties = new Properties();
-			PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-			map.from(this::getProtocol).to(properties.in(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG));
-			return properties;
 		}
 
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2021 the original author or authors.
+ * Copyright 2012-2022 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.