GH-3052: RetryableTopic SITRS default changes

Fixes: #3052 

* Change `RetryableTopic#SameIntervalTopicReuseStrategy` default behavior to SINGLE_TOPIC.
* Change unit test for SameIntervalTopicReuseStrategy.
* Add doc.

Author: Wzy19930507

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

@@ -16,8 +16,8 @@ IMPORTANT: You can set the `AckMode` mode you prefer, but `RECORD` is suggested.
 IMPORTANT: At this time this functionality doesn't support class level `@KafkaListener` annotations.
 
 When using a manual `AckMode` with `asyncAcks` set to true, the `DefaultErrorHandler` must be configured with `seekAfterError` set to `false`.
-Starting with versions 2.9.10, 3.0.8, this will be set to true unconditionally for such configurations.
-With earlier versions, it was necessary to override the `RetryConfigurationSupport.configureCustomizers()` method to set the property to `true`.
+Starting with versions 2.9.10, 3.0.8, this will be set to `false` unconditionally for such configurations.
+With earlier versions, it was necessary to override the `RetryConfigurationSupport.configureCustomizers()` method to set the property to `false`.
 
 [source, java]
 ----

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

@@ -111,35 +111,35 @@ This "final" retry topic will be suffixed with the provided or default suffix, a
 
 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.
 
-The default behavior is to work with the number of retry topics equal to the configured `maxAttempts` minus 1 and, when using exponential backoff, the retry topics are suffixed with the delay values, with the last retry topic (corresponding to the `maxInterval` delay) being suffixed with an additional index.
+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).
 
 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:
 
 * -retry-1000
 * -retry-2000
 * -retry-4000
 * -retry-8000
-* -retry-16000-0
-* -retry-16000-1
-* -retry-16000-2
-* ...
-* -retry-16000-224
+* -retry-16000
 
-When using the strategy that reuses the retry topic for the same intervals, in the same configuration above the needed retry topics would be:
+When using the strategy that work with the number of retry topics equal to the configured `maxAttempts` minus 1, the last retry topic (corresponding to the `maxInterval` delay) being suffixed with an additional index would be:
 
 * -retry-1000
 * -retry-2000
 * -retry-4000
 * -retry-8000
-* -retry-16000
+* -retry-16000-0
+* -retry-16000-1
+* -retry-16000-2
+* ...
+* -retry-16000-224
 
-This will be the default in a future release.
+If multiple topics are required, then that can be done using the following configuration.
 
 [source, java]
 ----
 @RetryableTopic(attempts = 230,
     backoff = @Backoff(delay = 1_000, multiplier = 2, maxDelay = 16_000),
-    sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+    sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
 @KafkaListener(topics = "my-annotated-topic")
 public void processMessage(MyPojo message) {
     // ... message processing

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -45,4 +45,9 @@ Deprecating the `transactionManager` property in `ContainerProperties` in favor
 === After Rollback Processing
 
 A new `AfterRollbackProcessor` API `processBatch` is provided.
-See xref:kafka/annotation-error-handling.adoc#after-rollback[After-rollback Processor] for more information.
+See xref:kafka/annotation-error-handling.adoc#after-rollback[After-rollback Processor] for more information.
+
+[[x32-retry-topic]]
+=== Change @RetryableTopic SameIntervalTopicReuseStrategy default value
+Change `@RetryableTopic` property `SameIntervalTopicReuseStrategy` default value to `SINGLE_TOPIC`.
+See xref:retrytopic/topic-naming.adoc#single-topic-maxinterval-delay[Single Topic for maxInterval Exponential Delay].

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

@@ -41,6 +41,8 @@
  * @author Fabio da Silva Jr.
  * @author João Lima
  * @author Adrian Chlebosz
+ * @author Wang Zhiyang
+ *
  * @since 2.7
  *
  * @see org.springframework.kafka.retrytopic.RetryTopicConfigurer
@@ -191,10 +193,12 @@
 
 	/**
 	 * Topic reuse strategy for sequential attempts made with a same backoff interval.
+	 * Starting 3.2, change default behavior to {@code SameIntervalTopicReuseStrategy.SINGLE_TOPIC}.
+	 *
 	 * @return the strategy.
 	 * @since 3.0.4
 	 */
-	SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy() default SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS;
+	SameIntervalTopicReuseStrategy sameIntervalTopicReuseStrategy() default SameIntervalTopicReuseStrategy.SINGLE_TOPIC;
 
 	/**
 	 * Whether or not create a DLT, and redeliver to the DLT if delivery fails or just give up.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-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.
@@ -65,6 +65,8 @@
 
 /**
  * @author Tomaz Fernandes
+ * @author Wang Zhiyang
+ *
  * @since 2.8.4
  */
 @SpringJUnitConfig
@@ -247,9 +249,7 @@ static class FrameworkFatalTopicListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@SuppressWarnings("deprecation")
-		@RetryableTopic(sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC,
-				backoff = @Backoff(50))
+		@RetryableTopic(backoff = @Backoff(50))
 		@KafkaListener(topics = FRAMEWORK_FATAL_EXCEPTION_TOPIC)
 		public void listenWithAnnotation(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
 			container.fatalFrameworkLatch.countDown();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-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.
@@ -384,6 +384,7 @@ static class FourthTopicListener {
 		CountDownLatchContainer container;
 
 		@RetryableTopic(dltStrategy = DltStrategy.NO_DLT, attempts = "4", backoff = @Backoff(300),
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS,
 				kafkaTemplate = "${kafka.template}")
 		@KafkaListener(topics = FOURTH_TOPIC, containerFactory = MAIN_TOPIC_CONTAINER_FACTORY)
 		public void listenNoDlt(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
@@ -410,6 +411,7 @@ static class FifthTopicListener1 {
 				numPartitions = "2",
 				retryTopicSuffix = "-listener1", dltTopicSuffix = "-listener1-dlt",
 				topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS,
 				kafkaTemplate = "${kafka.template}")
 		@KafkaListener(id = "fifthTopicId1", topicPartitions = {@TopicPartition(topic = TWO_LISTENERS_TOPIC,
 				partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0"))},
@@ -442,6 +444,7 @@ static class FifthTopicListener2 {
 				numPartitions = "2",
 				retryTopicSuffix = "-listener2", dltTopicSuffix = "-listener2-dlt",
 				topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS,
 				kafkaTemplate = "${kafka.template}")
 		@KafkaListener(id = "fifthTopicId2", topicPartitions = {@TopicPartition(topic = TWO_LISTENERS_TOPIC,
 				partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "0"))},
@@ -468,7 +471,8 @@ static class SixthTopicDefaultDLTListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@RetryableTopic(attempts = "4", backoff = @Backoff(50))
+		@RetryableTopic(attempts = "4", backoff = @Backoff(50),
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
 		@KafkaListener(id = "manual", topics = MANUAL_TOPIC, containerFactory = MAIN_TOPIC_CONTAINER_FACTORY)
 		public void listenNoDlt(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic,
 				@SuppressWarnings("unused") Acknowledgment ack) {
@@ -511,8 +515,7 @@ static class FirstReuseRetryTopicListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@RetryableTopic(attempts = "2", backoff = @Backoff(50),
-				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+		@RetryableTopic(attempts = "2", backoff = @Backoff(50))
 		@KafkaListener(id = "reuseRetry1", topics = FIRST_REUSE_RETRY_TOPIC,
 				containerFactory = "retryTopicListenerContainerFactory")
 		public void listen1(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
@@ -532,8 +535,7 @@ static class SecondReuseRetryTopicListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 30, maxDelay = 100, multiplier = 2),
-				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+		@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 30, maxDelay = 100, multiplier = 2))
 		@KafkaListener(id = "reuseRetry2", topics = SECOND_REUSE_RETRY_TOPIC,
 				containerFactory = "retryTopicListenerContainerFactory")
 		public void listen2(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
@@ -553,8 +555,7 @@ static class ThirdReuseRetryTopicListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 1, maxDelay = 5, multiplier = 1.4),
-				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+		@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 1, maxDelay = 5, multiplier = 1.4))
 		@KafkaListener(id = "reuseRetry3", topics = THIRD_REUSE_RETRY_TOPIC,
 				containerFactory = "retryTopicListenerContainerFactory")
 		public void listen3(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-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.
@@ -62,6 +62,8 @@
 /**
  * @author Tomaz Fernandes
  * @author Cenk Akin
+ * @author Wang Zhiyang
+ *
  * @since 2.8.3
  */
 @SpringJUnitConfig
@@ -121,7 +123,8 @@ static class FirstRetryableKafkaListener {
 				attempts = "4",
 				backoff = @Backoff(delay = 1000, multiplier = 2.0),
 				autoCreateTopics = "false",
-				topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE)
+				topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
 		@KafkaListener(topics = RetryTopicSameContainerFactoryIntegrationTests.FIRST_TOPIC)
 		public void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
 			countDownLatchContainer.countDownLatchFirstRetryable.countDown();
@@ -142,7 +145,7 @@ static class SecondRetryableKafkaListener {
 		@Autowired
 		CountDownLatchContainer countDownLatchContainer;
 
-		@RetryableTopic
+		@RetryableTopic(sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.MULTIPLE_TOPICS)
 		@KafkaListener(topics = RetryTopicSameContainerFactoryIntegrationTests.SECOND_TOPIC)
 		public void listen(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
 			countDownLatchContainer.countDownLatchSecondRetryable.countDown();

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

@@ -52,6 +52,8 @@
  * @author Tomaz Fernandes
  * @author Gary Russell
  * @author Adrian Chlebosz
+ * @author Wang Zhiyang
+ *
  * @since 2.7
  */
 @SuppressWarnings("deprecation")
@@ -314,15 +316,13 @@ void shouldCreateFixedBackoff() {
 
 		// then
 		List<DestinationTopic.Properties> destinationTopicProperties = configuration.getDestinationTopicProperties();
+		assertThat(destinationTopicProperties.size()).isEqualTo(3);
 		DestinationTopic destinationTopic = new DestinationTopic("", destinationTopicProperties.get(0));
 		assertThat(destinationTopic.getDestinationDelay()).isEqualTo(0);
 		DestinationTopic destinationTopic2 = new DestinationTopic("", destinationTopicProperties.get(1));
 		assertThat(destinationTopic2.getDestinationDelay()).isEqualTo(1000);
 		DestinationTopic destinationTopic3 = new DestinationTopic("", destinationTopicProperties.get(2));
-		assertThat(destinationTopic3.getDestinationDelay()).isEqualTo(1000);
-		DestinationTopic destinationTopic4 = new DestinationTopic("", destinationTopicProperties.get(3));
-		assertThat(destinationTopic4.getDestinationDelay()).isEqualTo(0);
-
+		assertThat(destinationTopic3.getDestinationDelay()).isEqualTo(0);
 	}
 
 	@Test