Merge pull request #34504 from Abhijeetmishr

* pr/34504:
  Polish "Switch to single topic for same intervals in Spring Kafka retry config"
  Switch to single topic for same intervals in Spring Kafka retry config

Closes gh-34504

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfiguration.java

@@ -157,12 +157,11 @@ public class KafkaAutoConfiguration {
 	@Bean
 	@ConditionalOnProperty(name = "spring.kafka.retry.topic.enabled")
 	@ConditionalOnSingleCandidate(KafkaTemplate.class)
-	@SuppressWarnings("deprecation")
 	public RetryTopicConfiguration kafkaRetryTopicConfiguration(KafkaTemplate<?, ?> kafkaTemplate) {
 		KafkaProperties.Retry.Topic retryTopic = this.properties.getRetry().getTopic();
 		RetryTopicConfigurationBuilder builder = RetryTopicConfigurationBuilder.newInstance()
 			.maxAttempts(retryTopic.getAttempts())
-			.useSingleTopicForFixedDelays()
+			.useSingleTopicForSameIntervals()
 			.suffixTopicsWithIndexValues()
 			.doNotAutoCreateRetryTopics();
 		setBackOffPolicy(builder, retryTopic);

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationIntegrationTests.java

@@ -105,7 +105,7 @@ class KafkaAutoConfigurationIntegrationTests {
 				"spring.kafka.consumer.auto-offset-reset=earliest");
 		RetryTopicConfiguration configuration = this.context.getBean(RetryTopicConfiguration.class);
 		assertThat(configuration.getDestinationTopicProperties()).extracting(DestinationTopic.Properties::delay)
-			.containsExactly(0L, 100L, 200L, 300L, 300L, 0L);
+			.containsExactly(0L, 100L, 200L, 300L, 0L);
 		KafkaTemplate<String, String> template = this.context.getBean(KafkaTemplate.class);
 		template.send(TEST_RETRY_TOPIC, "foo", "bar");
 		RetryListener listener = this.context.getBean(RetryListener.class);
@@ -116,7 +116,7 @@ class KafkaAutoConfigurationIntegrationTests {
 			.asList()
 			.hasSize(5)
 			.containsSequence("testRetryTopic", "testRetryTopic-retry-0", "testRetryTopic-retry-1",
-					"testRetryTopic-retry-2", "testRetryTopic-retry-3");
+					"testRetryTopic-retry-2");
 	}
 
 	@Test

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/kafka/KafkaAutoConfigurationTests.java

@@ -364,10 +364,10 @@ class KafkaAutoConfigurationTests {
 				"spring.kafka.retry.topic.multiplier=2", "spring.kafka.retry.topic.max-delay=300ms")
 			.run((context) -> {
 				RetryTopicConfiguration configuration = context.getBean(RetryTopicConfiguration.class);
-				assertThat(configuration.getDestinationTopicProperties()).hasSize(6)
+				assertThat(configuration.getDestinationTopicProperties()).hasSize(5)
 					.extracting(DestinationTopic.Properties::delay, DestinationTopic.Properties::suffix)
 					.containsExactly(tuple(0L, ""), tuple(100L, "-retry-0"), tuple(200L, "-retry-1"),
-							tuple(300L, "-retry-2"), tuple(300L, "-retry-3"), tuple(0L, "-dlt"));
+							tuple(300L, "-retry-2"), tuple(0L, "-dlt"));
 			});
 	}