GH-2432: Fix Retryable Topic Provisioning

Resolves #2432

Don't provision an individual retry topic bean, if there is already a `NewTopic`
bean with the same topic name.

**cherry-pick to 2.9.x, 2.8.x**

* Fix `TopicForRetry` removal logic; include `NewTopics` beans in logic.

* Improve test.
# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java
#	spring-kafka/src/main/java/org/springframework/kafka/retrytopic/RetryTopicConfigurer.java

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

@@ -22,13 +22,16 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
@@ -57,6 +60,7 @@
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
+import org.springframework.kafka.support.TopicForRetryable;
 
 /**
  * An admin that delegates to an {@link AdminClient} to create topics defined
@@ -178,10 +182,7 @@ public void afterSingletonsInstantiated() {
 	 * @see #setAutoCreate(boolean)
 	 */
 	public final boolean initialize() {
-		Collection<NewTopic> newTopics = new ArrayList<>(
-				this.applicationContext.getBeansOfType(NewTopic.class, false, false).values());
-		Collection<NewTopics> wrappers = this.applicationContext.getBeansOfType(NewTopics.class, false, false).values();
-		wrappers.forEach(wrapper -> newTopics.addAll(wrapper.getNewTopics()));
+		Collection<NewTopic> newTopics = newTopics();
 		if (newTopics.size() > 0) {
 			AdminClient adminClient = null;
 			try {
@@ -218,6 +219,41 @@ public final boolean initialize() {
 		return false;
 	}
 
+	/*
+	 * Remove any TopicForRetryable bean if there is also a NewTopic with the same topic name.
+	 */
+	private Collection<NewTopic> newTopics() {
+		Map<String, NewTopic> newTopicsMap = new HashMap<>(
+				this.applicationContext.getBeansOfType(NewTopic.class, false, false));
+		Map<String, NewTopics> wrappers = this.applicationContext.getBeansOfType(NewTopics.class, false, false);
+		AtomicInteger count = new AtomicInteger();
+		wrappers.forEach((name, newTopics) -> {
+			newTopics.getNewTopics().forEach(nt -> newTopicsMap.put(name + "#" + count.getAndIncrement(), nt));
+		});
+		Map<String, NewTopic> topicsForRetry = newTopicsMap.entrySet().stream()
+				.filter(entry -> entry.getValue() instanceof TopicForRetryable)
+				.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+		for (Entry<String, NewTopic> entry : topicsForRetry.entrySet()) {
+			Iterator<Entry<String, NewTopic>> iterator = newTopicsMap.entrySet().iterator();
+			boolean remove = false;
+			while (iterator.hasNext()) {
+				Entry<String, NewTopic> nt = iterator.next();
+				// if we have a NewTopic and TopicForRetry with the same name, remove the latter
+				if (nt.getValue().name().equals(entry.getValue().name())
+						&& !(nt.getValue() instanceof TopicForRetryable)) {
+
+					remove = true;
+					break;
+				}
+			}
+			if (remove) {
+				newTopicsMap.remove(entry.getKey());
+			}
+		}
+		Collection<NewTopic> newTopics = new ArrayList<>(newTopicsMap.values());
+		return newTopics;
+	}
+
 	@Override
 	public void createOrModifyTopics(NewTopic... topics) {
 		try (AdminClient client = createAdmin()) {

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

@@ -21,7 +21,6 @@
 import java.util.function.Consumer;
 
 import org.apache.commons.logging.LogFactory;
-import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 import org.springframework.beans.BeansException;
@@ -37,6 +36,7 @@
 import org.springframework.kafka.config.MultiMethodKafkaListenerEndpoint;
 import org.springframework.kafka.listener.ListenerUtils;
 import org.springframework.kafka.support.EndpointHandlerMethod;
+import org.springframework.kafka.support.TopicForRetryable;
 import org.springframework.lang.Nullable;
 
 
@@ -372,7 +372,7 @@ protected void createNewTopicBeans(Collection<String> topics, RetryTopicConfigur
 				String beanName = topic + "-topicRegistrationBean";
 				if (!bf.containsBean(beanName)) {
 					bf.registerSingleton(beanName,
-							new NewTopic(topic, config.getNumPartitions(), config.getReplicationFactor()));
+							new TopicForRetryable(topic, config.getNumPartitions(), config.getReplicationFactor()));
 				}
 			}
 		);

spring-kafka/src/main/java/org/springframework/kafka/support/TopicForRetryable.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2022 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.support;
+
+import org.apache.kafka.clients.admin.NewTopic;
+
+/**
+ * Marker to indicate this {@link NewTopic} is for retryable topics; admin will ignore these if
+ * a regular {@link NewTopic} exist.
+ *
+ * @author Gary Russell
+ * @since 2.8.10
+ *
+ */
+public class TopicForRetryable extends NewTopic {
+
+	/**
+	 * Create an instance with the provided properties.
+	 * @param topic the topic.
+	 * @param numPartitions the partitions.
+	 * @param replicationFactor the replication factor.
+	 */
+	public TopicForRetryable(String topic, int numPartitions, short replicationFactor) {
+		super(topic, numPartitions, replicationFactor);
+	}
+
+}

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

@@ -19,16 +19,22 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
 
+import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
@@ -47,10 +53,12 @@
 import org.springframework.kafka.annotation.RetryableTopic;
 import org.springframework.kafka.annotation.TopicPartition;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.config.TopicBuilder;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.kafka.core.KafkaAdmin.NewTopics;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ContainerProperties;
@@ -133,11 +141,30 @@ void shouldRetrySecondTopic() {
 	}
 
 	@Test
-	void shouldRetryThirdTopicWithTimeout() {
+	void shouldRetryThirdTopicWithTimeout(@Autowired KafkaAdmin admin) throws Exception {
 		logger.debug("Sending message to topic " + THIRD_TOPIC);
 		kafkaTemplate.send(THIRD_TOPIC, "Testing topic 3");
 		assertThat(awaitLatch(latchContainer.countDownLatch3)).isTrue();
 		assertThat(awaitLatch(latchContainer.countDownLatchDltOne)).isTrue();
+		Map<String, TopicDescription> topics = admin.describeTopics(THIRD_TOPIC, THIRD_TOPIC + "-dlt", FOURTH_TOPIC);
+		assertThat(topics.get(THIRD_TOPIC).partitions()).hasSize(2);
+		assertThat(topics.get(THIRD_TOPIC + "-dlt").partitions()).hasSize(3);
+		assertThat(topics.get(FOURTH_TOPIC).partitions()).hasSize(2);
+		AtomicReference<Method> method = new AtomicReference<>();
+		org.springframework.util.ReflectionUtils.doWithMethods(KafkaAdmin.class, m -> {
+			m.setAccessible(true);
+			method.set(m);
+		}, m -> m.getName().equals("newTopics"));
+		@SuppressWarnings("unchecked")
+		Collection<NewTopic> weededTopics = (Collection<NewTopic>) method.get().invoke(admin);
+		AtomicInteger weeded = new AtomicInteger();
+		weededTopics.forEach(topic -> {
+			if (topic.name().equals(THIRD_TOPIC) || topic.name().equals(FOURTH_TOPIC)) {
+				assertThat(topic).isExactlyInstanceOf(NewTopic.class);
+				weeded.incrementAndGet();
+			}
+		});
+		assertThat(weeded.get()).isEqualTo(2);
 	}
 
 	@Test
@@ -529,6 +556,16 @@ public KafkaAdmin kafkaAdmin() {
 			return new KafkaAdmin(configs);
 		}
 
+		@Bean
+		public NewTopic topic() {
+			return TopicBuilder.name(THIRD_TOPIC).partitions(2).replicas(1).build();
+		}
+
+		@Bean
+		public NewTopics topics() {
+			return new NewTopics(TopicBuilder.name(FOURTH_TOPIC).partitions(2).replicas(1).build());
+		}
+
 		@Bean
 		public ConsumerFactory<String, String> consumerFactory() {
 			Map<String, Object> props = new HashMap<>();