GH-1765 - Fix using same factory for retry & main (#1768)

Fixes #1765

Author: tomazfernandes

spring-kafka/src/main/java/org/springframework/kafka/listener/PartitionPausingBackoffManager.java

@@ -139,6 +139,7 @@ public PartitionPausingBackoffManager(ListenerContainerRegistry listenerContaine
 	@Override
 	public void backOffIfNecessary(Context context) {
 		long backoffTime = context.getDueTimestamp() - getCurrentMillisFromClock();
+		LOGGER.debug(() -> "Back off time: " + backoffTime + " Context: " + context);
 		if (backoffTime > 0) {
 			pauseConsumptionAndThrow(context, backoffTime);
 		}

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

@@ -98,10 +98,10 @@ public class ListenerContainerFactoryConfigurer {
 			ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory, Configuration configuration) {
 		return isCached(containerFactory)
 				? containerFactory
-				: doConfigure(containerFactory, configuration.backOffValues);
+				: addToCache(doConfigure(containerFactory, configuration.backOffValues));
 	}
 
-	public ConcurrentKafkaListenerContainerFactory<?, ?> configureWithoutBackOff(
+	public ConcurrentKafkaListenerContainerFactory<?, ?> configureWithoutBackOffValues(
 			ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory, Configuration configuration) {
 		return isCached(containerFactory)
 				? containerFactory
@@ -114,7 +114,7 @@ public class ListenerContainerFactoryConfigurer {
 				setupBackoffAwareMessageListenerAdapter(container, backOffValues));
 		containerFactory
 				.setErrorHandler(createErrorHandler(this.deadLetterPublishingRecovererFactory.create()));
-		return addToCache(containerFactory);
+		return containerFactory;
 	}
 
 	private boolean isCached(ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory) {

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

@@ -355,7 +355,7 @@ private EndpointHandlerMethod getDltEndpointHandlerMethodOrDefault(EndpointHandl
 				.resolveFactoryForMainEndpoint(providedFactory, defaultFactoryBeanName,
 						configuration.forContainerFactoryResolver());
 		return this.listenerContainerFactoryConfigurer
-				.configureWithoutBackOff(resolvedFactory, configuration.forContainerFactoryConfigurer());
+				.configureWithoutBackOffValues(resolvedFactory, configuration.forContainerFactoryConfigurer());
 	}
 
 	private ConcurrentKafkaListenerContainerFactory<?, ?> resolveAndConfigureFactoryForRetryEndpoint(

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

@@ -304,7 +304,7 @@ void shouldNotSetPolltimoutAndPartitionIdleIfNoBackOffProvided() {
 				new ListenerContainerFactoryConfigurer(kafkaConsumerBackoffManager,
 						deadLetterPublishingRecovererFactory, clock);
 		configurer
-				.configureWithoutBackOff(containerFactory, configuration.forContainerFactoryConfigurer());
+				.configureWithoutBackOffValues(containerFactory, configuration.forContainerFactoryConfigurer());
 
 		// then
 		then(containerFactory).should(times(1))

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

@@ -225,7 +225,7 @@ void shouldConfigureRetryEndpoints() {
 				defaultFactoryBeanName, factoryResolverConfig);
 		willReturn(containerFactory).given(this.listenerContainerFactoryConfigurer).configure(containerFactory,
 				lcfcConfiguration);
-		willReturn(containerFactory).given(this.listenerContainerFactoryConfigurer).configureWithoutBackOff(containerFactory,
+		willReturn(containerFactory).given(this.listenerContainerFactoryConfigurer).configureWithoutBackOffValues(containerFactory,
 				lcfcConfiguration);
 
 		RetryTopicConfigurer configurer = new RetryTopicConfigurer(destinationTopicProcessor, containerFactoryResolver,

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

@@ -72,14 +72,14 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
-@EmbeddedKafka(topics = { RetryableTopicIntegrationTests.FIRST_TOPIC,
-		RetryableTopicIntegrationTests.SECOND_TOPIC,
-		RetryableTopicIntegrationTests.THIRD_TOPIC,
-		RetryableTopicIntegrationTests.FOURTH_TOPIC }, partitions = 1)
+@EmbeddedKafka(topics = { RetryTopicIntegrationTests.FIRST_TOPIC,
+		RetryTopicIntegrationTests.SECOND_TOPIC,
+		RetryTopicIntegrationTests.THIRD_TOPIC,
+		RetryTopicIntegrationTests.FOURTH_TOPIC }, partitions = 1)
 @TestPropertySource(properties = "five.attempts=5")
-public class RetryableTopicIntegrationTests {
+public class RetryTopicIntegrationTests {
 
-	private static final Logger logger = LoggerFactory.getLogger(RetryableTopicIntegrationTests.class);
+	private static final Logger logger = LoggerFactory.getLogger(RetryTopicIntegrationTests.class);
 
 	public final static String FIRST_TOPIC = "myRetryTopic1";
 
@@ -183,7 +183,7 @@ static class ThirdTopicListener {
 		CountDownLatchContainer container;
 
 		@RetryableTopic(attempts = "${five.attempts}",
-				backoff = @Backoff(delay = 70, maxDelay = 120, multiplier = 1.5),
+				backoff = @Backoff(delay = 250, maxDelay = 1000, multiplier = 1.5),
 				numPartitions = "#{3}",
 				timeout = "${missing.property:2000}",
 				include = MyRetryException.class, kafkaTemplate = "kafkaTemplate")
@@ -207,7 +207,7 @@ static class FourthTopicListener {
 		@Autowired
 		CountDownLatchContainer container;
 
-		@RetryableTopic(dltStrategy = DltStrategy.NO_DLT, attempts = "4", backoff = @Backoff(50),
+		@RetryableTopic(dltStrategy = DltStrategy.NO_DLT, attempts = "4", backoff = @Backoff(300),
 				kafkaTemplate = "kafkaTemplate")
 		@KafkaListener(topics = FOURTH_TOPIC, containerFactory = MAIN_TOPIC_CONTAINER_FACTORY)
 		public void listenNoDlt(String message, @Header(KafkaHeaders.RECEIVED_TOPIC) String receivedTopic) {
@@ -320,7 +320,7 @@ public RetryTopicConfiguration firstRetryTopic(KafkaTemplate<String, String> tem
 		public RetryTopicConfiguration secondRetryTopic(KafkaTemplate<String, String> template) {
 			return RetryTopicConfigurationBuilder
 					.newInstance()
-					.exponentialBackoff(50, 2, 10000)
+					.exponentialBackoff(500, 2, 10000)
 					.retryOn(Arrays.asList(IllegalStateException.class, IllegalAccessException.class))
 					.traversingCauses()
 					.includeTopic(SECOND_TOPIC)
@@ -344,7 +344,6 @@ public ThirdTopicListener thirdTopicListener() {
 			return new ThirdTopicListener();
 		}
 
-
 		@Bean
 		public FourthTopicListener fourthTopicListener() {
 			return new FourthTopicListener();
@@ -482,10 +481,6 @@ public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerCont
 				ConsumerFactory<String, String> consumerFactory) {
 
 			ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
-			ContainerProperties props = factory.getContainerProperties();
-			props.setIdleEventInterval(100L);
-			props.setPollTimeout(50L);
-			props.setIdlePartitionEventInterval(100L);
 			factory.setConsumerFactory(consumerFactory);
 			factory.setConcurrency(1);
 			return factory;