Polish DestinationTopicPropertiesFactory and DestinationTopicPropertiesFactoryTests

Author: Wzy19930507

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java

@@ -16,10 +16,8 @@
 
 package org.springframework.kafka.retrytopic;
 
-import java.util.Arrays;
 import java.util.List;
 import java.util.function.BiPredicate;
-import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
 import org.springframework.classify.BinaryExceptionClassifier;
@@ -36,6 +34,7 @@
  * @author Tomaz Fernandes
  * @author Gary Russell
  * @author João Lima
+ * @author Wang Zhiyang
  * @since 2.7
  *
  */
@@ -47,19 +46,21 @@ public class DestinationTopicPropertiesFactory {
 
 	private final List<Long> backOffValues;
 
-	private final BinaryExceptionClassifier exceptionClassifier;
-
 	private final int numPartitions;
 
 	private final int maxAttempts;
 
-	private final KafkaOperations<?, ?> kafkaOperations;
+	private final boolean isSameIntervalReuse;
 
-	private final DltStrategy dltStrategy;
+	private final boolean isFixedDelay;
+
+	private final int retryTopicsAmount;
+
+	private final BiPredicate<Integer, Throwable> shouldRetryOn;
 
-	private final TopicSuffixingStrategy topicSuffixingStrategy;
+	private final KafkaOperations<?, ?> kafkaOperations;
 
-	private final SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy;
+	private final DltStrategy dltStrategy;
 
 	private final long timeout;
 
@@ -90,15 +91,19 @@ public DestinationTopicPropertiesFactory(String retryTopicSuffix, String dltSuff
 
 		this.dltStrategy = dltStrategy;
 		this.kafkaOperations = kafkaOperations;
-		this.exceptionClassifier = exceptionClassifier;
 		this.numPartitions = numPartitions;
-		this.topicSuffixingStrategy = topicSuffixingStrategy;
-		this.sameIntervalTopicReuseStrategy = sameIntervalTopicReuseStrategy;
 		this.timeout = timeout;
 		this.destinationTopicSuffixes = new DestinationTopicSuffixes(retryTopicSuffix, dltSuffix);
 		this.backOffValues = backOffValues;
-		// Max Attempts include the initial try.
-		this.maxAttempts = this.backOffValues.size() + 1;
+		// Max Attempts to include the initial try.
+		int backOffValuesSize = this.backOffValues.size();
+		this.maxAttempts = backOffValuesSize + 1;
+		this.shouldRetryOn = (attempt, throwable) -> attempt < this.maxAttempts
+				&& exceptionClassifier.classify(throwable);
+		this.isSameIntervalReuse = SameIntervalTopicReuseStrategy.SINGLE_TOPIC.equals(sameIntervalTopicReuseStrategy);
+		this.retryTopicsAmount = backOffValuesSize - reusableTopicAttempts();
+		this.isFixedDelay = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.equals(topicSuffixingStrategy)
+				|| backOffValuesSize > 1 && backOffValues.stream().distinct().count() == 1;
 	}
 
 	/**
@@ -113,71 +118,26 @@ public DestinationTopicPropertiesFactory autoStartDltHandler(@Nullable Boolean a
 	}
 
 	public List<DestinationTopic.Properties> createProperties() {
-		return isSingleTopicFixedDelay()
-				? createPropertiesForFixedDelaySingleTopic()
-				: createPropertiesForDefaultTopicStrategy();
-	}
-
-	private List<DestinationTopic.Properties> createPropertiesForFixedDelaySingleTopic() {
-		return isNoDltStrategy()
-					? Arrays.asList(createMainTopicProperties(),
-							createRetryProperties(1, getShouldRetryOn()))
-					: Arrays.asList(createMainTopicProperties(),
-							createRetryProperties(1, getShouldRetryOn()),
-							createDltProperties());
-	}
-
-	private boolean isSingleTopicFixedDelay() {
-		return (this.backOffValues.size() == 1 || isFixedDelay()) && isSingleTopicSameIntervalTopicReuseStrategy();
-	}
-
-	private boolean isSingleTopicSameIntervalTopicReuseStrategy() {
-		return SameIntervalTopicReuseStrategy.SINGLE_TOPIC.equals(this.sameIntervalTopicReuseStrategy);
-	}
-
-	private List<DestinationTopic.Properties> createPropertiesForDefaultTopicStrategy() {
-
-		int retryTopicsAmount = retryTopicsAmount();
-
-		return IntStream.rangeClosed(0, isNoDltStrategy()
-												? retryTopicsAmount
-												: retryTopicsAmount + 1)
-												.mapToObj(this::createTopicProperties)
-												.collect(Collectors.toList());
-	}
-
-	int retryTopicsAmount() {
-		return this.backOffValues.size() - reusableTopicAttempts();
-	}
-
-	private int reusableTopicAttempts() {
-		return this.backOffValues.size() > 0
-				? !isFixedDelay()
-					? isSingleTopicSameIntervalTopicReuseStrategy()
-						// Assuming that duplicates are always in
-						//  the end of the list.
-						? amountOfDuplicates(this.backOffValues.get(this.backOffValues.size() - 1)) - 1
-						: 0
-					: 0
-				: 0;
-	}
-
-	private boolean isNoDltStrategy() {
-		return DltStrategy.NO_DLT.equals(this.dltStrategy);
+		int topicAmount = DltStrategy.NO_DLT.equals(this.dltStrategy)
+				? this.retryTopicsAmount
+				: this.retryTopicsAmount + 1;
+		return IntStream
+				.rangeClosed(0, topicAmount)
+				.mapToObj(this::createTopicProperties)
+				.toList();
 	}
 
 	private DestinationTopic.Properties createTopicProperties(int index) {
-		BiPredicate<Integer, Throwable> shouldRetryOn = getShouldRetryOn();
 		return index == 0
 				? createMainTopicProperties()
-				: (index <= this.retryTopicsAmount())
-					? createRetryProperties(index, shouldRetryOn)
+				: index <= this.retryTopicsAmount
+					? createRetryProperties(index)
 					: createDltProperties();
 	}
 
 	private DestinationTopic.Properties createMainTopicProperties() {
 		return new DestinationTopic.Properties(0, MAIN_TOPIC_SUFFIX, DestinationTopic.Type.MAIN, this.maxAttempts,
-				this.numPartitions, this.dltStrategy, this.kafkaOperations, getShouldRetryOn(), this.timeout);
+				this.numPartitions, this.dltStrategy, this.kafkaOperations, this.shouldRetryOn, this.timeout);
 	}
 
 	private DestinationTopic.Properties createDltProperties() {
@@ -186,49 +146,42 @@ private DestinationTopic.Properties createDltProperties() {
 				this.kafkaOperations, (a, e) -> false, this.timeout, this.autoStartDltHandler);
 	}
 
-	private BiPredicate<Integer, Throwable> getShouldRetryOn() {
-		return (attempt, throwable) -> attempt < this.maxAttempts && this.exceptionClassifier.classify(throwable);
-	}
-
-	private DestinationTopic.Properties createRetryProperties(int index,
-			BiPredicate<Integer, Throwable> shouldRetryOn) {
-
+	private DestinationTopic.Properties createRetryProperties(int index) {
 		int indexInBackoffValues = index - 1;
-		Long thisBackOffValue = this.backOffValues.get(indexInBackoffValues);
-		DestinationTopic.Type topicTypeToUse = isDelayWithReusedTopic(thisBackOffValue)
-				? Type.REUSABLE_RETRY_TOPIC
-				: Type.RETRY;
-		return createProperties(topicTypeToUse, shouldRetryOn, indexInBackoffValues,
-				getTopicSuffix(indexInBackoffValues, thisBackOffValue));
-	}
-
-	private String getTopicSuffix(int indexInBackoffValues, Long thisBackOffValue) {
-		return isSingleTopicFixedDelay()
-				? this.destinationTopicSuffixes.getRetrySuffix()
-				: isSuffixWithIndexStrategy() || isFixedDelay()
-					? joinWithRetrySuffix(indexInBackoffValues)
-					: hasDuplicates(thisBackOffValue)
-						? joinWithRetrySuffix(thisBackOffValue)
-							.concat(suffixForRepeatedInterval(indexInBackoffValues, thisBackOffValue))
-						: joinWithRetrySuffix(thisBackOffValue);
-	}
-
-	private String suffixForRepeatedInterval(int indexInBackoffValues, Long thisBackOffValue) {
-		return isSingleTopicSameIntervalTopicReuseStrategy()
-				? ""
-				: "-" + getIndexInBackoffValues(indexInBackoffValues, thisBackOffValue);
+		long thisBackOffValue = this.backOffValues.get(indexInBackoffValues);
+		return createProperties(thisBackOffValue, getTopicSuffix(indexInBackoffValues, thisBackOffValue));
 	}
 
-	private boolean isDelayWithReusedTopic(Long backoffValue) {
-		return hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy();
+	private String getTopicSuffix(int indexInBackoffValues, long thisBackOffValue) {
+		if (this.isSameIntervalReuse && this.retryTopicsAmount == 1) {
+			return this.destinationTopicSuffixes.getRetrySuffix();
+		}
+		else if (this.isFixedDelay) {
+			return joinWithRetrySuffix(indexInBackoffValues);
+		}
+		else {
+			String retrySuffix = joinWithRetrySuffix(thisBackOffValue);
+			if (!this.isSameIntervalReuse && hasDuplicates(thisBackOffValue)) {
+				return retrySuffix.concat("-" + getIndexInBackoffValues(indexInBackoffValues, thisBackOffValue));
+			}
+			return retrySuffix;
+		}
 	}
 
 	private int getIndexInBackoffValues(int indexInBackoffValues, Long thisBackOffValue) {
 		return indexInBackoffValues - this.backOffValues.indexOf(thisBackOffValue);
 	}
 
-	private boolean isSuffixWithIndexStrategy() {
-		return TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.equals(this.topicSuffixingStrategy);
+	private DestinationTopic.Type getDestinationTopicType(Long backOffValue) {
+		return this.isSameIntervalReuse && hasDuplicates(backOffValue) ? Type.REUSABLE_RETRY_TOPIC : Type.RETRY;
+	}
+
+	private int reusableTopicAttempts() {
+		if (this.isSameIntervalReuse && this.backOffValues.size() > 1) {
+			// Assuming that duplicates are always at the end of the list.
+			return amountOfDuplicates(this.backOffValues.get(this.backOffValues.size() - 1)) - 1;
+		}
+		return 0;
 	}
 
 	private boolean hasDuplicates(Long thisBackOffValue) {
@@ -238,22 +191,15 @@ private boolean hasDuplicates(Long thisBackOffValue) {
 	private int amountOfDuplicates(Long thisBackOffValue) {
 		return Long.valueOf(this.backOffValues
 				.stream()
-				.filter(value -> value.equals(thisBackOffValue))
-				.count()).intValue();
-	}
-
-	private DestinationTopic.Properties createProperties(DestinationTopic.Type topicType,
-														BiPredicate<Integer, Throwable> shouldRetryOn,
-														int indexInBackoffValues,
-														String suffix) {
-		return new DestinationTopic.Properties(this.backOffValues.get(indexInBackoffValues), suffix,
-				topicType, this.maxAttempts, this.numPartitions, this.dltStrategy,
-				this.kafkaOperations, shouldRetryOn, this.timeout);
+				.filter(thisBackOffValue::equals)
+				.count())
+				.intValue();
 	}
 
-	private boolean isFixedDelay() {
-		// If all values are the same, such as in NoBackOffPolicy and FixedBackoffPolicy
-		return this.backOffValues.size() > 1 && this.backOffValues.stream().distinct().count() == 1;
+	private DestinationTopic.Properties createProperties(long delayMs, String suffix) {
+		return new DestinationTopic.Properties(delayMs, suffix, getDestinationTopicType(delayMs),
+				this.maxAttempts, this.numPartitions, this.dltStrategy, this.kafkaOperations, this.shouldRetryOn,
+				this.timeout);
 	}
 
 	private String joinWithRetrySuffix(long parameter) {