Set RetryTopicConfigurationBuilder default to reuse same topic for identical intervals

Author: bky373

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/how-the-pattern-works.adoc

@@ -6,6 +6,8 @@ The retry topic consumer then checks the timestamp and if it's not due it pauses
 When it is due the partition consumption is resumed, and the message is consumed again.
 If the message processing fails again the message will be forwarded to the next retry topic, and the pattern is repeated until a successful processing occurs, or the attempts are exhausted, and the message is sent to the Dead Letter Topic (if configured).
 
+NOTE: Starting with version 3.2, the default behavior is to reuse the retry topic for the same intervals.
+
 To illustrate, if you have a "main-topic" topic, and want to set up non-blocking retry with an exponential backoff of 1000ms with a multiplier of 2 and 4 max attempts, it will create the main-topic-retry-1000, main-topic-retry-2000, main-topic-retry-4000 and main-topic-dlt topics and configure the respective consumers.
 The framework also takes care of creating the topics and setting up and configuring the listeners.
 

spring-kafka-docs/src/main/antora/modules/ROOT/pages/retrytopic/topic-naming.adoc

@@ -9,8 +9,8 @@ Examples:
 
 "my-other-topic" -> "my-topic-myRetrySuffix-1000", "my-topic-myRetrySuffix-2000", ..., "my-topic-myDltSuffix"
 
-NOTE: The default behavior is to create separate retry topics for each attempt, appended with an index value: retry-0, retry-1, ..., retry-n.
-Therefore, by default the number of retry topics is the configured `maxAttempts` minus 1.
+NOTE: Starting with version 3.2, the default behavior is to reuse the retry topic for the same intervals. Alternatively, you can configure separate retry topics for each attempt, appended with an index(or delay) value: retry-0, retry-1, ..., retry-n, by using `SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS`.
+If you choose this strategy, the number of retry topics will, by default, be `maxAttempts` minus 1.
 
 You can xref:retrytopic/topic-naming.adoc#retry-topics-and-dlt-suffixes[configure the suffixes], choose whether to append xref:retrytopic/topic-naming.adoc#append-index-or-delay[the attempt index or delay], use a xref:retrytopic/topic-naming.adoc#single-topic-fixed-delay[single retry topic when using fixed backoff], and use a xref:retrytopic/topic-naming.adoc#single-topic-maxinterval-delay[single retry topic for the attempts with the maxInterval] when using exponential backoffs.
 
@@ -69,8 +69,8 @@ public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> templa
 
 NOTE: The default behavior is to suffix with the delay values, except for fixed delay configurations with multiple topics, in which case the topics are suffixed with the topic's index.
 
-[[single-topic-fixed-delay]]
-== Single Topic for Fixed Delay Retries
+[[topic-strategies-for-fixed-delay]]
+== Topic Strategies for Fixed Delay Retries
 
 If you're using fixed delay policies such as `FixedBackOffPolicy` or `NoBackOffPolicy` you can use a single topic to accomplish the non-blocking retries.
 This topic will be suffixed with the provided or default suffix, and will not have either the index or the delay values appended.
@@ -79,39 +79,64 @@ NOTE: The previous `FixedDelayStrategy` is now deprecated, and can be replaced b
 
 [source, java]
 ----
-@RetryableTopic(backoff = @Backoff(2_000), fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)
+@RetryableTopic(
+        backoff = @Backoff(2_000),
+        sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC // default
+)
 @KafkaListener(topics = "my-annotated-topic")
 public void processMessage(MyPojo message) {
     // ... message processing
 }
 ----
 
-[source, java]
+[source,java]
 ----
 @Bean
 public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
     return RetryTopicConfigurationBuilder
             .newInstance()
-            .fixedBackOff(3_000)
-            .maxAttempts(5)
-            .useSingleTopicForFixedDelays()
+            .fixedBackOff(2_000)
+            .sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy.SINGLE_TOPIC) // default
             .create(template);
 }
 ----
 
-NOTE: The default behavior is creating separate retry topics for each attempt, appended with their index values: retry-0, retry-1, ...
+If multiple topics are required, then that can be done using the following configuration.
 
+[source, java]
+----
+@RetryableTopic(
+        backoff = @Backoff(3_000),
+        sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS
+)
+@KafkaListener(topics = "my-annotated-topic")
+public void processMessage(MyPojo message) {
+    // ... message processing
+}
+----
 
-[[single-topic-maxinterval-delay]]
-== Single Topic for maxInterval Exponential Delay 
+[source,java]
+----
+@Bean
+public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
+    return RetryTopicConfigurationBuilder
+            .newInstance()
+            .fixedBackOff(3_000)
+            .sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
+            .create(template);
+}
+----
+
+[[topic-strategies-for-maxinterval-delay]]
+== Topic Strategies for maxInterval Exponential Delay
 
 If you're using exponential backoff policy (`ExponentialBackOffPolicy`), you can use a single retry topic to accomplish the non-blocking retries of the attempts whose delays are the configured `maxInterval`.
 
 This "final" retry topic will be suffixed with the provided or default suffix, and will have either the index or the `maxInterval` value appended.
 
 NOTE: By opting to use a single topic for the retries with the `maxInterval` delay, it may become more viable to configure an exponential retry policy that keeps retrying for a long time, because in this approach you do not need a large amount of topics.
 
-Starting 3.2, the default behavior is reuses the retry topic for the same intervals, when using exponential backoff, the retry topics are suffixed with the delay values, with the last retry topic reuses for the same intervals(corresponding to the `maxInterval` delay).
+Starting with version 3.2, the default behavior is to reuse the retry topic for the same intervals. When using exponential backoff, the retry topics are suffixed with the delay values, with the last retry topic being reused for the same intervals(corresponding to the `maxInterval` delay).
 
 For instance, when configuring the exponential backoff with `initialInterval=1_000`, `multiplier=2`, and `maxInterval=16_000`, in order to keep trying for one hour, one would need to configure `maxAttempts` as 229, and by default the needed retry topics would be:
 
@@ -151,10 +176,10 @@ public void processMessage(MyPojo message) {
 @Bean
 public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, MyPojo> template) {
     return RetryTopicConfigurationBuilder
-            .newInstance()            
+            .newInstance()
             .exponentialBackoff(1_000, 2, 16_000)
             .maxAttempts(230)
-            .useSingleTopicForSameIntervals()
+            .sameIntervalTopicReuseStrategy(SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
             .create(template);
 }
 ----

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2024 the original author or authors.
+ * Copyright 2018-2025 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.
@@ -46,6 +46,7 @@
  * @author Gary Russell
  * @author Adrian Chlebosz
  * @author Wang Zhiyang
+ * @author Borahm Lee
  *
  * @since 2.7
  *
@@ -88,7 +89,7 @@ public class RetryTopicConfigurationBuilder {
 
 	private TopicSuffixingStrategy topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_DELAY_VALUE;
 
-	private SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS;
+	private SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC;
 
 	@Nullable
 	private Boolean autoStartDltHandler;
@@ -255,10 +256,6 @@ 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 #useSingleTopicForSameIntervals()}.
 	 * @param sameIntervalTopicReuseStrategy the strategy.
 	 * @return the builder.
 	 * @since 3.0.4
@@ -279,6 +276,7 @@ public RetryTopicConfigurationBuilder sameIntervalTopicReuseStrategy(SameInterva
 	 * @since 3.0.4
 	 * @see SameIntervalTopicReuseStrategy
 	 */
+	@Deprecated(since = "3.4", forRemoval = true)
 	public RetryTopicConfigurationBuilder useSingleTopicForSameIntervals() {
 		this.sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC;
 		return this;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2024-2025 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.
@@ -86,6 +86,7 @@
 
 /**
  * @author Sanghyeok An
+ * @author Borahm Lee
  * @since 3.3.0
  */
 
@@ -738,7 +739,6 @@ RetryTopicConfiguration firstRetryTopic(KafkaTemplate<String, String> template)
 					.fixedBackOff(50)
 					.maxAttempts(5)
 					.concurrency(1)
-					.useSingleTopicForSameIntervals()
 					.includeTopic(FIRST_TOPIC)
 					.doNotRetryOnDltFailure()
 					.dltHandlerMethod("myCustomDltProcessor", DLT_METHOD_NAME)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2024-2025 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.
@@ -66,6 +66,7 @@
 /**
  * @author Sanghyeok An
  * @author Artem Bilan
+ * @author Borahm Lee
  *
  * @since 3.3.0
  */
@@ -1113,7 +1114,6 @@ static  RetryTopicConfiguration createRetryTopicConfiguration(
 					.fixedBackOff(50)
 					.maxAttempts(maxAttempts)
 					.concurrency(1)
-					.useSingleTopicForSameIntervals()
 					.includeTopic(topicName)
 					.doNotRetryOnDltFailure()
 					.dltHandlerMethod(dltBeanName, DLT_METHOD_NAME)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2024-2025 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.
@@ -86,6 +86,7 @@
 
 /**
  * @author Sanghyeok An
+ * @author Borahm Lee
  * @since 3.3.0
  */
 
@@ -731,7 +732,6 @@ RetryTopicConfiguration firstRetryTopic(KafkaTemplate<String, String> template)
 					.fixedBackOff(50)
 					.maxAttempts(5)
 					.concurrency(1)
-					.useSingleTopicForSameIntervals()
 					.includeTopic(FIRST_TOPIC)
 					.doNotRetryOnDltFailure()
 					.dltHandlerMethod("myCustomDltProcessor", DLT_METHOD_NAME)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2024 the original author or authors.
+ * Copyright 2024-2025 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.
@@ -64,6 +64,7 @@
 
 /**
  * @author Sanghyeok An
+ * @author Borahm Lee
  * @since 3.3.0
  */
 
@@ -1108,7 +1109,6 @@ static  RetryTopicConfiguration createRetryTopicConfiguration(
 					.fixedBackOff(50)
 					.maxAttempts(maxAttempts)
 					.concurrency(1)
-					.useSingleTopicForSameIntervals()
 					.includeTopic(topicName)
 					.doNotRetryOnDltFailure()
 					.dltHandlerMethod(dltBeanName, DLT_METHOD_NAME)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2024 the original author or authors.
+ * Copyright 2021-2025 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.
@@ -95,6 +95,7 @@
  * @author Wang Zhiyang
  * @author Artem Bilan
  * @author Sanghyeok An
+ * @author Borahm Lee
  *
  * @since 3.2
  */
@@ -625,7 +626,6 @@ RetryTopicConfiguration firstRetryTopic(KafkaTemplate<String, String> template)
 					.fixedBackOff(50)
 					.maxAttempts(5)
 					.concurrency(1)
-					.useSingleTopicForSameIntervals()
 					.includeTopic(FIRST_TOPIC)
 					.doNotRetryOnDltFailure()
 					.dltHandlerMethod("myCustomDltProcessor", DLT_METHOD_NAME)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2024 the original author or authors.
+ * Copyright 2018-2025 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.
@@ -37,6 +37,7 @@
 /**
  * @author Tomaz Fernandes
  * @author Adrian Chlebosz
+ * @author Borahm Lee
  * @since 2.7
  */
 @ExtendWith(MockitoExtension.class)
@@ -81,9 +82,10 @@ void shouldSetFixedBackOffPolicy() {
 		// then
 		List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
 		assertThat(destinationTopicProperties.get(0).delay()).isEqualTo(0);
+		assertThat(destinationTopicProperties.get(1).isRetryTopic()).isTrue();
 		assertThat(destinationTopicProperties.get(1).delay()).isEqualTo(1000);
-		assertThat(destinationTopicProperties.get(2).delay()).isEqualTo(1000);
-		assertThat(destinationTopicProperties.get(3).delay()).isEqualTo(0);
+		assertThat(destinationTopicProperties.get(2).isDltTopic()).isTrue();
+		assertThat(destinationTopicProperties.get(2).delay()).isEqualTo(0);
 	}
 
 	@Test
@@ -99,11 +101,10 @@ void shouldSetNoBackoffPolicy() {
 		// then
 		List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
 		assertThat(destinationTopicProperties.get(0).delay()).isEqualTo(0);
+		assertThat(destinationTopicProperties.get(1).isRetryTopic()).isTrue();
 		assertThat(destinationTopicProperties.get(1).delay()).isEqualTo(0);
+		assertThat(destinationTopicProperties.get(2).isDltTopic()).isTrue();
 		assertThat(destinationTopicProperties.get(2).delay()).isEqualTo(0);
-		assertThat(destinationTopicProperties.get(3).delay()).isEqualTo(0);
-
-
 	}
 
 	@Test
@@ -203,7 +204,7 @@ void shouldSetDltRoutingRules() {
 			.create(kafkaOperations);
 
 		// then
-		DestinationTopic.Properties desExcDltProps = configuration.getDestinationTopicProperties().get(3);
+		DestinationTopic.Properties desExcDltProps = configuration.getDestinationTopicProperties().get(2);
 		assertThat(desExcDltProps.suffix()).isEqualTo("-deserialization-dlt");
 		assertThat(desExcDltProps.usedForExceptions()).containsExactly(DeserializationException.class);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2024 the original author or authors.
+ * Copyright 2021-2025 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.
@@ -95,6 +95,7 @@
  * @author Gary Russell
  * @author Wang Zhiyang
  * @author Sanghyeok An
+ * @author Borahm Lee
  * @since 2.7
  */
 @SpringJUnitConfig
@@ -652,7 +653,6 @@ public RetryTopicConfiguration firstRetryTopic(KafkaTemplate<String, String> tem
 					.fixedBackOff(50)
 					.maxAttempts(5)
 					.concurrency(1)
-					.useSingleTopicForSameIntervals()
 					.includeTopic(FIRST_TOPIC)
 					.doNotRetryOnDltFailure()
 					.dltHandlerMethod("myCustomDltProcessor", DLT_METHOD_NAME)