Skip to content

Commit 69f201d

Browse files
authored
DestinationTopicPropertiesFactory refactoring
* Refactor `DestinationTopicPropertiesFactory` and `DestinationTopicPropertiesFactoryTests` * remove method ambiguity in `getIndexInBackoffValues()` and cleanup method `createProperties()`
1 parent 8a684b3 commit 69f201d

File tree

2 files changed

+81
-161
lines changed

2 files changed

+81
-161
lines changed

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

Lines changed: 62 additions & 127 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2018-2023 the original author or authors.
2+
* Copyright 2018-2024 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,11 +16,10 @@
1616

1717
package org.springframework.kafka.retrytopic;
1818

19-
import java.util.Arrays;
19+
import java.util.ArrayList;
20+
import java.util.Collections;
2021
import java.util.List;
2122
import java.util.function.BiPredicate;
22-
import java.util.stream.Collectors;
23-
import java.util.stream.IntStream;
2423

2524
import org.springframework.classify.BinaryExceptionClassifier;
2625
import org.springframework.kafka.core.KafkaOperations;
@@ -36,6 +35,7 @@
3635
* @author Tomaz Fernandes
3736
* @author Gary Russell
3837
* @author João Lima
38+
* @author Wang Zhiyang
3939
* @since 2.7
4040
*
4141
*/
@@ -47,19 +47,21 @@ public class DestinationTopicPropertiesFactory {
4747

4848
private final List<Long> backOffValues;
4949

50-
private final BinaryExceptionClassifier exceptionClassifier;
51-
5250
private final int numPartitions;
5351

5452
private final int maxAttempts;
5553

56-
private final KafkaOperations<?, ?> kafkaOperations;
54+
private final boolean isSameIntervalReuse;
5755

58-
private final DltStrategy dltStrategy;
56+
private final boolean isFixedDelay;
5957

60-
private final TopicSuffixingStrategy topicSuffixingStrategy;
58+
private final int retryTopicsAmount;
6159

62-
private final SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy;
60+
private final BiPredicate<Integer, Throwable> shouldRetryOn;
61+
62+
private final KafkaOperations<?, ?> kafkaOperations;
63+
64+
private final DltStrategy dltStrategy;
6365

6466
private final long timeout;
6567

@@ -90,15 +92,19 @@ public DestinationTopicPropertiesFactory(String retryTopicSuffix, String dltSuff
9092

9193
this.dltStrategy = dltStrategy;
9294
this.kafkaOperations = kafkaOperations;
93-
this.exceptionClassifier = exceptionClassifier;
9495
this.numPartitions = numPartitions;
95-
this.topicSuffixingStrategy = topicSuffixingStrategy;
96-
this.sameIntervalTopicReuseStrategy = sameIntervalTopicReuseStrategy;
9796
this.timeout = timeout;
9897
this.destinationTopicSuffixes = new DestinationTopicSuffixes(retryTopicSuffix, dltSuffix);
9998
this.backOffValues = backOffValues;
100-
// Max Attempts include the initial try.
101-
this.maxAttempts = this.backOffValues.size() + 1;
99+
int backOffValuesSize = this.backOffValues.size();
100+
this.isSameIntervalReuse = SameIntervalTopicReuseStrategy.SINGLE_TOPIC.equals(sameIntervalTopicReuseStrategy);
101+
this.isFixedDelay = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.equals(topicSuffixingStrategy)
102+
|| backOffValuesSize > 1 && backOffValues.stream().distinct().count() == 1;
103+
// Max Attempts to include the initial try.
104+
this.maxAttempts = backOffValuesSize + 1;
105+
this.shouldRetryOn = (attempt, throwable) -> attempt < this.maxAttempts
106+
&& exceptionClassifier.classify(throwable);
107+
this.retryTopicsAmount = backOffValuesSize - reusableTopicAttempts();
102108
}
103109

104110
/**
@@ -113,71 +119,20 @@ public DestinationTopicPropertiesFactory autoStartDltHandler(@Nullable Boolean a
113119
}
114120

115121
public List<DestinationTopic.Properties> createProperties() {
116-
return isSingleTopicFixedDelay()
117-
? createPropertiesForFixedDelaySingleTopic()
118-
: createPropertiesForDefaultTopicStrategy();
119-
}
120-
121-
private List<DestinationTopic.Properties> createPropertiesForFixedDelaySingleTopic() {
122-
return isNoDltStrategy()
123-
? Arrays.asList(createMainTopicProperties(),
124-
createRetryProperties(1, getShouldRetryOn()))
125-
: Arrays.asList(createMainTopicProperties(),
126-
createRetryProperties(1, getShouldRetryOn()),
127-
createDltProperties());
128-
}
129-
130-
private boolean isSingleTopicFixedDelay() {
131-
return (this.backOffValues.size() == 1 || isFixedDelay()) && isSingleTopicSameIntervalTopicReuseStrategy();
132-
}
133-
134-
private boolean isSingleTopicSameIntervalTopicReuseStrategy() {
135-
return SameIntervalTopicReuseStrategy.SINGLE_TOPIC.equals(this.sameIntervalTopicReuseStrategy);
136-
}
137-
138-
private List<DestinationTopic.Properties> createPropertiesForDefaultTopicStrategy() {
139-
140-
int retryTopicsAmount = retryTopicsAmount();
141-
142-
return IntStream.rangeClosed(0, isNoDltStrategy()
143-
? retryTopicsAmount
144-
: retryTopicsAmount + 1)
145-
.mapToObj(this::createTopicProperties)
146-
.collect(Collectors.toList());
147-
}
148-
149-
int retryTopicsAmount() {
150-
return this.backOffValues.size() - reusableTopicAttempts();
151-
}
152-
153-
private int reusableTopicAttempts() {
154-
return this.backOffValues.size() > 0
155-
? !isFixedDelay()
156-
? isSingleTopicSameIntervalTopicReuseStrategy()
157-
// Assuming that duplicates are always in
158-
// the end of the list.
159-
? amountOfDuplicates(this.backOffValues.get(this.backOffValues.size() - 1)) - 1
160-
: 0
161-
: 0
162-
: 0;
163-
}
164-
165-
private boolean isNoDltStrategy() {
166-
return DltStrategy.NO_DLT.equals(this.dltStrategy);
167-
}
168-
169-
private DestinationTopic.Properties createTopicProperties(int index) {
170-
BiPredicate<Integer, Throwable> shouldRetryOn = getShouldRetryOn();
171-
return index == 0
172-
? createMainTopicProperties()
173-
: (index <= this.retryTopicsAmount())
174-
? createRetryProperties(index, shouldRetryOn)
175-
: createDltProperties();
122+
List<DestinationTopic.Properties> list = new ArrayList<>(this.retryTopicsAmount + 2);
123+
list.add(createMainTopicProperties());
124+
for (int backOffIndex = 0; backOffIndex < this.retryTopicsAmount; backOffIndex++) {
125+
list.add(createRetryProperties(backOffIndex));
126+
}
127+
if (!DltStrategy.NO_DLT.equals(this.dltStrategy)) {
128+
list.add(createDltProperties());
129+
}
130+
return Collections.unmodifiableList(list);
176131
}
177132

178133
private DestinationTopic.Properties createMainTopicProperties() {
179134
return new DestinationTopic.Properties(0, MAIN_TOPIC_SUFFIX, DestinationTopic.Type.MAIN, this.maxAttempts,
180-
this.numPartitions, this.dltStrategy, this.kafkaOperations, getShouldRetryOn(), this.timeout);
135+
this.numPartitions, this.dltStrategy, this.kafkaOperations, this.shouldRetryOn, this.timeout);
181136
}
182137

183138
private DestinationTopic.Properties createDltProperties() {
@@ -186,49 +141,37 @@ private DestinationTopic.Properties createDltProperties() {
186141
this.kafkaOperations, (a, e) -> false, this.timeout, this.autoStartDltHandler);
187142
}
188143

189-
private BiPredicate<Integer, Throwable> getShouldRetryOn() {
190-
return (attempt, throwable) -> attempt < this.maxAttempts && this.exceptionClassifier.classify(throwable);
144+
private DestinationTopic.Properties createRetryProperties(int backOffIndex) {
145+
long thisBackOffValue = this.backOffValues.get(backOffIndex);
146+
return createProperties(thisBackOffValue, getTopicSuffix(backOffIndex, thisBackOffValue));
191147
}
192148

193-
private DestinationTopic.Properties createRetryProperties(int index,
194-
BiPredicate<Integer, Throwable> shouldRetryOn) {
195-
196-
int indexInBackoffValues = index - 1;
197-
Long thisBackOffValue = this.backOffValues.get(indexInBackoffValues);
198-
DestinationTopic.Type topicTypeToUse = isDelayWithReusedTopic(thisBackOffValue)
199-
? Type.REUSABLE_RETRY_TOPIC
200-
: Type.RETRY;
201-
return createProperties(topicTypeToUse, shouldRetryOn, indexInBackoffValues,
202-
getTopicSuffix(indexInBackoffValues, thisBackOffValue));
203-
}
204-
205-
private String getTopicSuffix(int indexInBackoffValues, Long thisBackOffValue) {
206-
return isSingleTopicFixedDelay()
207-
? this.destinationTopicSuffixes.getRetrySuffix()
208-
: isSuffixWithIndexStrategy() || isFixedDelay()
209-
? joinWithRetrySuffix(indexInBackoffValues)
210-
: hasDuplicates(thisBackOffValue)
211-
? joinWithRetrySuffix(thisBackOffValue)
212-
.concat(suffixForRepeatedInterval(indexInBackoffValues, thisBackOffValue))
213-
: joinWithRetrySuffix(thisBackOffValue);
214-
}
215-
216-
private String suffixForRepeatedInterval(int indexInBackoffValues, Long thisBackOffValue) {
217-
return isSingleTopicSameIntervalTopicReuseStrategy()
218-
? ""
219-
: "-" + getIndexInBackoffValues(indexInBackoffValues, thisBackOffValue);
220-
}
221-
222-
private boolean isDelayWithReusedTopic(Long backoffValue) {
223-
return hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy();
149+
private String getTopicSuffix(int backOffIndex, long thisBackOffValue) {
150+
if (this.isSameIntervalReuse && this.retryTopicsAmount == 1) {
151+
return this.destinationTopicSuffixes.getRetrySuffix();
152+
}
153+
else if (this.isFixedDelay) {
154+
return joinWithRetrySuffix(backOffIndex);
155+
}
156+
else {
157+
String retrySuffix = joinWithRetrySuffix(thisBackOffValue);
158+
if (!this.isSameIntervalReuse && hasDuplicates(thisBackOffValue)) {
159+
return retrySuffix.concat("-" + (backOffIndex - this.backOffValues.indexOf(thisBackOffValue)));
160+
}
161+
return retrySuffix;
162+
}
224163
}
225164

226-
private int getIndexInBackoffValues(int indexInBackoffValues, Long thisBackOffValue) {
227-
return indexInBackoffValues - this.backOffValues.indexOf(thisBackOffValue);
165+
private DestinationTopic.Type getDestinationTopicType(Long backOffValue) {
166+
return this.isSameIntervalReuse && hasDuplicates(backOffValue) ? Type.REUSABLE_RETRY_TOPIC : Type.RETRY;
228167
}
229168

230-
private boolean isSuffixWithIndexStrategy() {
231-
return TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.equals(this.topicSuffixingStrategy);
169+
private int reusableTopicAttempts() {
170+
if (this.isSameIntervalReuse && this.backOffValues.size() > 1) {
171+
// Assuming that duplicates are always at the end of the list.
172+
return amountOfDuplicates(this.backOffValues.get(this.backOffValues.size() - 1)) - 1;
173+
}
174+
return 0;
232175
}
233176

234177
private boolean hasDuplicates(Long thisBackOffValue) {
@@ -238,22 +181,14 @@ private boolean hasDuplicates(Long thisBackOffValue) {
238181
private int amountOfDuplicates(Long thisBackOffValue) {
239182
return Long.valueOf(this.backOffValues
240183
.stream()
241-
.filter(value -> value.equals(thisBackOffValue))
242-
.count()).intValue();
243-
}
244-
245-
private DestinationTopic.Properties createProperties(DestinationTopic.Type topicType,
246-
BiPredicate<Integer, Throwable> shouldRetryOn,
247-
int indexInBackoffValues,
248-
String suffix) {
249-
return new DestinationTopic.Properties(this.backOffValues.get(indexInBackoffValues), suffix,
250-
topicType, this.maxAttempts, this.numPartitions, this.dltStrategy,
251-
this.kafkaOperations, shouldRetryOn, this.timeout);
184+
.filter(thisBackOffValue::equals)
185+
.count())
186+
.intValue();
252187
}
253188

254-
private boolean isFixedDelay() {
255-
// If all values are the same, such as in NoBackOffPolicy and FixedBackoffPolicy
256-
return this.backOffValues.size() > 1 && this.backOffValues.stream().distinct().count() == 1;
189+
private DestinationTopic.Properties createProperties(long delayMs, String suffix) {
190+
return new DestinationTopic.Properties(delayMs, suffix, getDestinationTopicType(delayMs), this.maxAttempts,
191+
this.numPartitions, this.dltStrategy, this.kafkaOperations, this.shouldRetryOn, this.timeout);
257192
}
258193

259194
private String joinWithRetrySuffix(long parameter) {

0 commit comments

Comments
 (0)