GH-2654: Incorrect retry topic suffix (#2876)

* GH-2654: Incorrect retry topic suffix

Resolves #2654

* fix incorrect retry topic suffix when attempts is 2

* optimization method `isDelayWithReusedTopic`

* add retry topic integration tests when reuse retry topic

fix java doc in RetryTopic

* fix javadoc in `@RetryTopic`

* fix javadoc in `RetryTopicConfigurationProvider`

* fix javadoc in `RetryableTopic`

* add a note to the `whats-new.adoc`

* Doc polishing.

---------

Co-authored-by: Zhiyang.Wang1 <[email protected]>
Co-authored-by: Gary Russell <[email protected]>

Author: 3 people

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

@@ -38,3 +38,9 @@ See xref:kafka/container-factory.adoc[Container Factory] for more information.
 You can now add a `Validator` to this deserializer; if the delegate `Deserializer` successfully deserializes the object, but that object fails validation, an exception is thrown similar to a deserialization exception occurring.
 This allows the original raw data to be passed to the error handler.
 See xref:kafka/serdes.adoc#error-handling-deserializer[Using `ErrorHandlingDeserializer`] for more information.
+
+[[x31-retryable]]
+=== Retryable Topics
+Change suffix `-retry-5000` to `-retry` when `@RetryableTopic(backoff = @Backoff(delay = 5000), attempts = "2", fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC)`.
+If you want to keep suffix `-retry-5000`, use `@RetryableTopic(backoff = @Backoff(delay = 5000), attempts = "2")`.
+See xref:retrytopic/topic-naming.adoc[Topic Naming] for more information.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -43,7 +43,7 @@
  * annotation is available.
  *
  * If beans are found in the container there's a check to determine whether or not the
- * provided topics topics should be handled by any of such instances.
+ * provided topics should be handled by any of such instances.
  *
  * If the annotation is provided, a
  * {@link org.springframework.kafka.annotation.DltHandler} annotated method is looked up.

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

@@ -128,7 +128,7 @@ private List<DestinationTopic.Properties> createPropertiesForFixedDelaySingleTop
 	}
 
 	private boolean isSingleTopicFixedDelay() {
-		return isFixedDelay() && isSingleTopicSameIntervalTopicReuseStrategy();
+		return (this.backOffValues.size() == 1 || isFixedDelay()) && isSingleTopicSameIntervalTopicReuseStrategy();
 	}
 
 	private boolean isSingleTopicSameIntervalTopicReuseStrategy() {
@@ -220,8 +220,7 @@ private String suffixForRepeatedInterval(int indexInBackoffValues, Long thisBack
 	}
 
 	private boolean isDelayWithReusedTopic(Long backoffValue) {
-		return ((isSingleTopicFixedDelay()) ||
-				(hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy()));
+		return hasDuplicates(backoffValue) && isSingleTopicSameIntervalTopicReuseStrategy();
 	}
 
 	private int getIndexInBackoffValues(int indexInBackoffValues, Long thisBackOffValue) {

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

@@ -93,6 +93,7 @@
 /**
  * @author Tomaz Fernandes
  * @author Gary Russell
+ * @author Wang Zhiyang
  * @since 2.7
  */
 @SpringJUnitConfig
@@ -122,6 +123,12 @@ public class RetryTopicIntegrationTests {
 
 	public final static String NOT_RETRYABLE_EXCEPTION_TOPIC = "noRetryTopic";
 
+	public final static String FIRST_REUSE_RETRY_TOPIC = "reuseRetry1";
+
+	public final static String SECOND_REUSE_RETRY_TOPIC = "reuseRetry2";
+
+	public final static String THIRD_REUSE_RETRY_TOPIC = "reuseRetry3";
+
 	private final static String MAIN_TOPIC_CONTAINER_FACTORY = "kafkaListenerContainerFactory";
 
 	@Autowired
@@ -267,6 +274,29 @@ void shouldRetryManualTopicWithDefaultDlt(@Autowired KafkaListenerEndpointRegist
 		}
 	}
 
+	@Test
+	void shouldFirstReuseRetryTopic(@Autowired FirstReuseRetryTopicListener listener1,
+			@Autowired SecondReuseRetryTopicListener listener2, @Autowired ThirdReuseRetryTopicListener listener3) {
+
+		logger.debug("Sending message to topic " + FIRST_REUSE_RETRY_TOPIC);
+		kafkaTemplate.send(FIRST_REUSE_RETRY_TOPIC, "Testing reuse topic 1");
+		logger.debug("Sending message to topic " + SECOND_REUSE_RETRY_TOPIC);
+		kafkaTemplate.send(SECOND_REUSE_RETRY_TOPIC, "Testing reuse topic 2");
+		logger.debug("Sending message to topic " + THIRD_REUSE_RETRY_TOPIC);
+		kafkaTemplate.send(THIRD_REUSE_RETRY_TOPIC, "Testing reuse topic 3");
+		assertThat(awaitLatch(latchContainer.countDownLatchReuseOne)).isTrue();
+		assertThat(awaitLatch(latchContainer.countDownLatchReuseTwo)).isTrue();
+		assertThat(awaitLatch(latchContainer.countDownLatchReuseThree)).isTrue();
+		assertThat(listener1.topics).containsExactly(FIRST_REUSE_RETRY_TOPIC,
+				FIRST_REUSE_RETRY_TOPIC + "-retry");
+		assertThat(listener2.topics).containsExactly(SECOND_REUSE_RETRY_TOPIC,
+				SECOND_REUSE_RETRY_TOPIC + "-retry-30", SECOND_REUSE_RETRY_TOPIC + "-retry-60",
+				SECOND_REUSE_RETRY_TOPIC + "-retry-100", SECOND_REUSE_RETRY_TOPIC + "-retry-100");
+		assertThat(listener3.topics).containsExactly(THIRD_REUSE_RETRY_TOPIC,
+				THIRD_REUSE_RETRY_TOPIC + "-retry", THIRD_REUSE_RETRY_TOPIC + "-retry",
+				THIRD_REUSE_RETRY_TOPIC + "-retry", THIRD_REUSE_RETRY_TOPIC + "-retry");
+	}
+
 	@Test
 	public void shouldGoStraightToDlt() {
 		logger.debug("Sending message to topic " + NOT_RETRYABLE_EXCEPTION_TOPIC);
@@ -473,6 +503,69 @@ public void annotatedDltMethod(Object message) {
 		}
 	}
 
+	@Component
+	static class FirstReuseRetryTopicListener {
+
+		final List<String> topics = Collections.synchronizedList(new ArrayList<>());
+
+		@Autowired
+		CountDownLatchContainer container;
+
+		@RetryableTopic(attempts = "2", backoff = @Backoff(50),
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+		@KafkaListener(id = "reuseRetry1", topics = FIRST_REUSE_RETRY_TOPIC,
+				containerFactory = "retryTopicListenerContainerFactory")
+		public void listen1(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
+			logger.debug("Message {} received in topic {} ", message, receivedTopic);
+			this.topics.add(receivedTopic);
+			container.countDownLatchReuseOne.countDown();
+			throw new RuntimeException("Another woooops... " + receivedTopic);
+		}
+
+	}
+
+	@Component
+	static class SecondReuseRetryTopicListener {
+
+		final List<String> topics = Collections.synchronizedList(new ArrayList<>());
+
+		@Autowired
+		CountDownLatchContainer container;
+
+		@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 30, maxDelay = 100, multiplier = 2),
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+		@KafkaListener(id = "reuseRetry2", topics = SECOND_REUSE_RETRY_TOPIC,
+				containerFactory = "retryTopicListenerContainerFactory")
+		public void listen2(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
+			logger.debug("Message {} received in topic {} ", message, receivedTopic);
+			this.topics.add(receivedTopic);
+			container.countDownLatchReuseTwo.countDown();
+			throw new RuntimeException("Another woooops... " + receivedTopic);
+		}
+
+	}
+
+	@Component
+	static class ThirdReuseRetryTopicListener {
+
+		final List<String> topics = Collections.synchronizedList(new ArrayList<>());
+
+		@Autowired
+		CountDownLatchContainer container;
+
+		@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 1, maxDelay = 5, multiplier = 1.4),
+				sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC)
+		@KafkaListener(id = "reuseRetry3", topics = THIRD_REUSE_RETRY_TOPIC,
+				containerFactory = "retryTopicListenerContainerFactory")
+		public void listen3(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
+			logger.debug("Message {} received in topic {} ", message, receivedTopic);
+			this.topics.add(receivedTopic);
+			container.countDownLatchReuseThree.countDown();
+			throw new RuntimeException("Another woooops... " + receivedTopic);
+		}
+
+	}
+
 	@Component
 	static class CountDownLatchContainer {
 
@@ -488,6 +581,9 @@ static class CountDownLatchContainer {
 		CountDownLatch countDownLatchDltTwo = new CountDownLatch(1);
 		CountDownLatch countDownLatchDltThree = new CountDownLatch(1);
 		CountDownLatch countDownLatchDltFour = new CountDownLatch(1);
+		CountDownLatch countDownLatchReuseOne = new CountDownLatch(2);
+		CountDownLatch countDownLatchReuseTwo = new CountDownLatch(5);
+		CountDownLatch countDownLatchReuseThree = new CountDownLatch(5);
 		CountDownLatch customDltCountdownLatch = new CountDownLatch(1);
 		CountDownLatch customErrorHandlerCountdownLatch = new CountDownLatch(6);
 		CountDownLatch customMessageConverterCountdownLatch = new CountDownLatch(6);
@@ -628,6 +724,21 @@ public NoRetryTopicListener noRetryTopicListener() {
 			return new NoRetryTopicListener();
 		}
 
+		@Bean
+		public FirstReuseRetryTopicListener firstReuseRetryTopicListener() {
+			return new FirstReuseRetryTopicListener();
+		}
+
+		@Bean
+		public SecondReuseRetryTopicListener secondReuseRetryTopicListener() {
+			return new SecondReuseRetryTopicListener();
+		}
+
+		@Bean
+		public ThirdReuseRetryTopicListener thirdReuseRetryTopicListener() {
+			return new ThirdReuseRetryTopicListener();
+		}
+
 		@Bean
 		CountDownLatchContainer latchContainer() {
 			return new CountDownLatchContainer();