Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Polish DestinationTopicPropertiesFactory #3004

Merged
merged 3 commits into from
Jan 30, 2024
Merged

Polish DestinationTopicPropertiesFactory #3004

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1a9d63d
Polish `DestinationTopicPropertiesFactory` and `DestinationTopicPrope…
Wzy19930507 Jan 26, 2024
c85a8ba
change copyright year to 2024
Wzy19930507 Jan 27, 2024
598322a
remove method `getIndexInBackoffValues` cause ambiguity and polish me…
Wzy19930507 Jan 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
189 changes: 62 additions & 127 deletions ...src/main/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2018-2023 the original author or authors.
* Copyright 2018-2024 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.
Expand All @@ -16,11 +16,10 @@

package org.springframework.kafka.retrytopic;

import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.function.BiPredicate;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import org.springframework.classify.BinaryExceptionClassifier;
import org.springframework.kafka.core.KafkaOperations;
Expand All @@ -36,6 +35,7 @@
* @author Tomaz Fernandes
* @author Gary Russell
* @author João Lima
* @author Wang Zhiyang
* @since 2.7
*
*/
Expand All @@ -47,19 +47,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 TopicSuffixingStrategy topicSuffixingStrategy;
private final int retryTopicsAmount;

private final SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy;
private final BiPredicate<Integer, Throwable> shouldRetryOn;

private final KafkaOperations<?, ?> kafkaOperations;

private final DltStrategy dltStrategy;

private final long timeout;

Expand Down Expand Up @@ -90,15 +92,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;
int backOffValuesSize = this.backOffValues.size();
this.isSameIntervalReuse = SameIntervalTopicReuseStrategy.SINGLE_TOPIC.equals(sameIntervalTopicReuseStrategy);
this.isFixedDelay = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.equals(topicSuffixingStrategy)
|| backOffValuesSize > 1 && backOffValues.stream().distinct().count() == 1;
// Max Attempts to include the initial try.
this.maxAttempts = backOffValuesSize + 1;
this.shouldRetryOn = (attempt, throwable) -> attempt < this.maxAttempts
&& exceptionClassifier.classify(throwable);
this.retryTopicsAmount = backOffValuesSize - reusableTopicAttempts();
}

/**
Expand All @@ -113,71 +119,20 @@ 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);
}

private DestinationTopic.Properties createTopicProperties(int index) {
BiPredicate<Integer, Throwable> shouldRetryOn = getShouldRetryOn();
return index == 0
? createMainTopicProperties()
: (index <= this.retryTopicsAmount())
? createRetryProperties(index, shouldRetryOn)
: createDltProperties();
List<DestinationTopic.Properties> list = new ArrayList<>(this.retryTopicsAmount + 2);
list.add(createMainTopicProperties());
for (int backOffIndex = 0; backOffIndex < this.retryTopicsAmount; backOffIndex++) {
list.add(createRetryProperties(backOffIndex));
}
if (!DltStrategy.NO_DLT.equals(this.dltStrategy)) {
list.add(createDltProperties());
}
return Collections.unmodifiableList(list);
}

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() {
Expand All @@ -186,49 +141,37 @@ 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 backOffIndex) {
long thisBackOffValue = this.backOffValues.get(backOffIndex);
return createProperties(thisBackOffValue, getTopicSuffix(backOffIndex, thisBackOffValue));
}

private DestinationTopic.Properties createRetryProperties(int index,
BiPredicate<Integer, Throwable> shouldRetryOn) {

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);
}

private boolean isDelayWithReusedTopic(Long backoffValue) {
return hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy();
private String getTopicSuffix(int backOffIndex, long thisBackOffValue) {
if (this.isSameIntervalReuse && this.retryTopicsAmount == 1) {
return this.destinationTopicSuffixes.getRetrySuffix();
}
else if (this.isFixedDelay) {
return joinWithRetrySuffix(backOffIndex);
}
else {
String retrySuffix = joinWithRetrySuffix(thisBackOffValue);
if (!this.isSameIntervalReuse && hasDuplicates(thisBackOffValue)) {
return retrySuffix.concat("-" + (backOffIndex - this.backOffValues.indexOf(thisBackOffValue)));
}
return retrySuffix;
}
}

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

private boolean isSuffixWithIndexStrategy() {
return TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE.equals(this.topicSuffixingStrategy);
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) {
Expand All @@ -238,22 +181,14 @@ 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) {
Expand Down
Loading
Loading