GH-2496: Reuse retry topic for maxInterval delay

spring-kafka/src/main/java/org/springframework/kafka/annotation/RetryableTopic.java

@@ -39,6 +39,7 @@
  * @author Tomaz Fernandes
  * @author Gary Russell
  * @author Fabio da Silva Jr.
+ * @author João Lima
  * @since 2.7
  *
  * @see org.springframework.kafka.retrytopic.RetryTopicConfigurer
@@ -181,6 +182,10 @@
 
 	/**
 	 * Topic reuse strategy for sequential attempts made with a same backoff interval.
+	 *
+	 * <p>Note: for fixed backoffs, when this is configured as
+	 * {@link SameIntervalTopicReuseStrategy#SINGLE_TOPIC}, it has precedence over
+	 * the configuration in {@link #fixedDelayTopicStrategy()}.
 	 * @return the strategy.
 	 */
 	SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy() default SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS;
@@ -194,9 +199,9 @@
 	/**
 	 * Whether to use a single or multiple topics when using a fixed delay.
 	 * @return the fixed delay strategy.
-	 * @deprecated in a future release, will be replaced by {@link #sameIntervalTopicReuseStrategy()}.
+	 * @deprecated in favor of {@link #sameIntervalTopicReuseStrategy()}.
 	 */
-	@Deprecated(forRemoval = true) // in 3.1
+	@Deprecated
 	FixedDelayStrategy fixedDelayTopicStrategy() default FixedDelayStrategy.MULTIPLE_TOPICS;
 
 	/**

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

@@ -72,7 +72,14 @@ public boolean isReusableRetryTopic() {
 		return Type.REUSABLE_RETRY_TOPIC.equals(this.properties.type);
 	}
 
-	@Deprecated(forRemoval = true) // in 3.1
+	/**
+	 * Whether this is a single retry topic.
+	 *
+	 * @return whether this is a single retry topic.
+	 * @deprecated in favor of using {@link DestinationTopic.Type#REUSABLE_RETRY_TOPIC}
+	 * and {@link #isReusableRetryTopic()}.
+	 */
+	@Deprecated
 	public boolean isSingleTopicRetry() {
 		return Type.SINGLE_TOPIC_RETRY.equals(this.properties.type);
 	}

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

@@ -129,7 +129,7 @@ private List<DestinationTopic.Properties> createPropertiesForFixedDelaySingleTop
 	}
 
 	private boolean isSingleTopicFixedDelay() {
-		return isFixedDelay() && isSingleTopicStrategy();
+		return isFixedDelay() && (isSingleTopicStrategy() || isSingleTopicSameIntervalTopicReuseStrategy());
 	}
 
 	private boolean isSingleTopicStrategy() {

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

@@ -22,10 +22,10 @@
  *
  * @author Tomaz Fernandes
  * @since 2.7
- * @deprecated in a future release, will be replaced by {@link SameIntervalTopicReuseStrategy}.
+ * @deprecated in favor of {@link SameIntervalTopicReuseStrategy}.
  *
  */
-@Deprecated(forRemoval = true) // in 3.1
+@Deprecated
 public enum FixedDelayStrategy {
 
 	/**

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

@@ -245,6 +245,11 @@ public RetryTopicConfigurationBuilder setTopicSuffixingStrategy(TopicSuffixingSt
 
 	/**
 	 * Configure the {@link SameIntervalTopicReuseStrategy}.
+	 *
+	 * <p>Note: for fixed backoffs, when this is configured as
+	 * {@link SameIntervalTopicReuseStrategy#SINGLE_TOPIC}, it has precedence over
+	 * the configuration done through
+	 * {@link #useSingleTopicForFixedDelays(FixedDelayStrategy)}.
 	 * @param sameIntervalTopicReuseStrategy the strategy.
 	 * @return the builder.
 	 */
@@ -404,10 +409,10 @@ public RetryTopicConfigurationBuilder fixedBackOff(int interval) {
 	/**
 	 * Configure the use of a single retry topic with fixed delays.
 	 * @return the builder.
-	 * @deprecated in a future release, configuration for single retry topic with fixed delays will have to be done with {@link #useSingleTopicForSameIntervals()}.
+	 * @deprecated in favor of {@link #useSingleTopicForSameIntervals()}.
 	 * @see FixedDelayStrategy#SINGLE_TOPIC
 	 */
-	@Deprecated(forRemoval = true) // in 3.1
+	@Deprecated
 	public RetryTopicConfigurationBuilder useSingleTopicForFixedDelays() {
 		this.fixedDelayStrategy = FixedDelayStrategy.SINGLE_TOPIC;
 		return this;
@@ -418,9 +423,10 @@ public RetryTopicConfigurationBuilder useSingleTopicForFixedDelays() {
 	 * {@link FixedDelayStrategy#MULTIPLE_TOPICS}.
 	 * @param delayStrategy the delay strategy.
 	 * @return the builder.
-	 * @deprecated in a future release, retry topic reuse configuration for fixed delays will have to be done with {@link #sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy)}.
+	 * @deprecated in favor of
+	 * {@link #sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy)}.
 	 */
-	@Deprecated(forRemoval = true) // in 3.1
+	@Deprecated
 	public RetryTopicConfigurationBuilder useSingleTopicForFixedDelays(FixedDelayStrategy delayStrategy) {
 		this.fixedDelayStrategy = delayStrategy;
 		return this;

spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DestinationTopicPropertiesFactoryTests.java

@@ -238,6 +238,48 @@ void shouldCreateOneRetryPropertyForFixedBackoffWithSingleTopicStrategy() {
 		assertThat(dltTopic.getDestinationPartitions()).isEqualTo(numPartitions);
 	}
 
+	@Test
+	void shouldCreateOneRetryPropertyForFixedBackoffWithSingleTopicSameIntervalReuseStrategy() {
+
+		// when
+		FixedBackOffPolicy backOffPolicy = new FixedBackOffPolicy();
+		backOffPolicy.setBackOffPeriod(1000);
+		int maxAttempts = 5;
+
+		List<Long> backOffValues = new BackOffValuesGenerator(maxAttempts, backOffPolicy).generateValues();
+
+		List<DestinationTopic.Properties> propertiesList =
+				new DestinationTopicPropertiesFactory(retryTopicSuffix, dltSuffix, backOffValues,
+						classifier, numPartitions, kafkaOperations, FixedDelayStrategy.MULTIPLE_TOPICS,
+						dltStrategy, suffixWithDelayValueSuffixingStrategy, singleTopicSameIntervalReuseStrategy,
+						-1).createProperties();
+
+		List<DestinationTopic> destinationTopicList = propertiesList
+				.stream()
+				.map(properties -> new DestinationTopic("mainTopic" + properties.suffix(), properties))
+				.collect(Collectors.toList());
+
+		// then
+		assertThat(propertiesList.size() == 3).isTrue();
+
+		DestinationTopic mainDestinationTopic = destinationTopicList.get(0);
+		assertThat(mainDestinationTopic.isMainTopic()).isTrue();
+
+		DestinationTopic.Properties firstRetryProperties = propertiesList.get(1);
+		assertThat(firstRetryProperties.suffix()).isEqualTo(retryTopicSuffix);
+		DestinationTopic retryDestinationTopic = destinationTopicList.get(1);
+		assertThat(retryDestinationTopic.isSingleTopicRetry()).isTrue();
+		assertThat(retryDestinationTopic.isReusableRetryTopic()).isFalse();
+		assertThat(retryDestinationTopic.getDestinationDelay()).isEqualTo(1000);
+
+		DestinationTopic.Properties dltProperties = propertiesList.get(2);
+		assertThat(dltProperties.suffix()).isEqualTo(dltSuffix);
+		assertThat(dltProperties.isDltTopic()).isTrue();
+		DestinationTopic dltTopic = destinationTopicList.get(2);
+		assertThat(dltTopic.getDestinationDelay()).isEqualTo(0);
+		assertThat(dltTopic.getDestinationPartitions()).isEqualTo(numPartitions);
+	}
+
 	@Test
 	void shouldCreateRetryPropertiesForFixedBackoffWithMultiTopicStrategy() {