Refactor newTopics to remove nested loops

Replace O(n²) nested iteration with O(n) map-based lookups for filtering topics
Use topicNameToMapKey to efficiently resolve duplicate topic names
This improves application startup time and scalability when managing
a large number of topics, while preserving the original filtering logic

Resolves: #4025

Signed-off-by: Choi Wang Gyu <[email protected]>

Author: cwangg897

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

@@ -22,7 +22,6 @@
 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;
@@ -333,31 +332,25 @@ protected Collection<NewTopic> newTopics() {
 		Map<String, NewTopic> topicsForRetry = newTopicsMap.entrySet().stream()
 				.filter(entry -> entry.getValue() instanceof TopicForRetryable)
 				.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+		Map<String, String> topicNameToMapKey = new HashMap<>();
+		for (Entry<String, NewTopic> entry : newTopicsMap.entrySet()) {
+			topicNameToMapKey.put(entry.getValue().name(), entry.getKey());
+		}
+
 		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;
+			String retryTopicName = entry.getValue().name();
+			String keyInNewTopicsMap = topicNameToMapKey.get(retryTopicName);
+			if (keyInNewTopicsMap != null) {
+				NewTopic existing = newTopicsMap.get(keyInNewTopicsMap);
+				if (!(existing instanceof TopicForRetryable)) {
+					newTopicsMap.remove(keyInNewTopicsMap);
 				}
 			}
-			if (remove) {
-				newTopicsMap.remove(entry.getKey());
-			}
 		}
-		Iterator<Entry<String, NewTopic>> iterator = newTopicsMap.entrySet().iterator();
-		while (iterator.hasNext()) {
-			Entry<String, NewTopic> next = iterator.next();
-			if (!this.createOrModifyTopic.test(next.getValue())) {
-				iterator.remove();
-			}
-		}
-		return new ArrayList<>(newTopicsMap.values());
+		Map<String, NewTopic> filteredMap = newTopicsMap.entrySet().stream()
+				.filter(entry -> this.createOrModifyTopic.test(entry.getValue()))
+				.collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+		return new ArrayList<>(filteredMap.values());
 	}
 
 	@Override