GH-2720: Add Predicate to KafkaAdmin

Resolves #2720

**cherry-pick to 2.9.x**
# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/KafkaAdmin.java

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -152,6 +152,16 @@ private KafkaAdmin admin;
 ----
 ====
 
+Starting with versions 2.9.10, 3.0.9, you can provide a `Predicate<NewTopic>` which can be used to determine whether a particular `NewTopic` bean should be considered for creation or modification.
+This is useful, for example, if you have multiple `KafkaAdmin` instances pointing to different clusters and you wish to select those topics that should be created or modified by each admin.
+
+====
+[source, java]
+----
+admin.setCreateOrModifyTopic(nt -> !nt.name().equals("dontCreateThisOne"));
+----
+====
+
 [[sending-messages]]
 ==== Sending Messages
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-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.
@@ -32,6 +32,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
@@ -61,6 +62,7 @@
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.support.TopicForRetryable;
+import org.springframework.util.Assert;
 
 /**
  * An admin that delegates to an {@link AdminClient} to create topics defined
@@ -87,6 +89,8 @@ public class KafkaAdmin extends KafkaResourceFactory
 
 	private ApplicationContext applicationContext;
 
+	private Predicate<NewTopic> createOrModifyTopic = nt -> true;
+
 	private Duration closeTimeout = DEFAULT_CLOSE_TIMEOUT;
 
 	private int operationTimeout = DEFAULT_OPERATION_TIMEOUT;
@@ -157,6 +161,31 @@ public void setModifyTopicConfigs(boolean modifyTopicConfigs) {
 		this.modifyTopicConfigs = modifyTopicConfigs;
 	}
 
+	/**
+	 * Set a predicate that returns true if a discovered {@link NewTopic} bean should be
+	 * considered for creation or modification by this admin instance. The default
+	 * predicate returns true for all {@link NewTopic}s. Used by the default
+	 * implementation of {@link #newTopics()}.
+	 * @param createOrModifyTopic the predicate.
+	 * @since 2.9.10
+	 * @see #newTopics()
+	 */
+	public void setCreateOrModifyTopic(Predicate<NewTopic> createOrModifyTopic) {
+		Assert.notNull(createOrModifyTopic, "'createOrModifyTopic' cannot be null");
+		this.createOrModifyTopic = createOrModifyTopic;
+	}
+
+	/**
+	 * Return the predicate used to determine whether a {@link NewTopic} should be
+	 * considered for creation or modification.
+	 * @return the predicate.
+	 * @since 2.9.10
+	 * @see #newTopics()
+	 */
+	protected Predicate<NewTopic> getCreateOrModifyTopic() {
+		return this.createOrModifyTopic;
+	}
+
 	@Override
 	public Map<String, Object> getConfigurationProperties() {
 		Map<String, Object> configs2 = new HashMap<>(this.configs);
@@ -219,10 +248,17 @@ public final boolean initialize() {
 		return false;
 	}
 
-	/*
-	 * Remove any TopicForRetryable bean if there is also a NewTopic with the same topic name.
+	/**
+	 * Return a collection of {@link NewTopic}s to create or modify. The default
+	 * implementation retrieves all {@link NewTopic} beans in the application context and
+	 * applies the {@link #setCreateOrModifyTopic(Predicate)} predicate to each one. It
+	 * also removes any {@link TopicForRetryable} bean if there is also a NewTopic with
+	 * the same topic name. This is performed before calling the predicate.
+	 * @return the collection of {@link NewTopic}s.
+	 * @since 2.9.10
+	 * @see #setCreateOrModifyTopic(Predicate)
 	 */
-	private Collection<NewTopic> newTopics() {
+	protected 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);
@@ -250,6 +286,13 @@ private Collection<NewTopic> newTopics() {
 				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());
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaAdminTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-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.
@@ -85,6 +85,9 @@ public class KafkaAdminTests {
 	@Autowired
 	private NewTopic mismatchconfig;
 
+	@Autowired
+	private NewTopic dontCreateThisOne;
+
 	@Test
 	public void testTopicConfigs() {
 		assertThat(topic1.configs()).containsEntry(
@@ -97,6 +100,7 @@ public void testTopicConfigs() {
 					.replicas(3)
 					.build().replicationFactor()).isEqualTo((short) 3);
 		assertThat(topic3.replicasAssignments()).hasSize(3);
+		assertThat(admin.newTopics()).doesNotContain(this.dontCreateThisOne);
 	}
 
 	@Test
@@ -269,6 +273,7 @@ public KafkaAdmin admin() {
 			KafkaAdmin admin = new KafkaAdmin(configs);
 			admin.setBootstrapServersSupplier(() ->
 					StringUtils.arrayToCommaDelimitedString(kafkaEmbedded().getBrokerAddresses()));
+			admin.setCreateOrModifyTopic(nt -> !nt.name().equals("dontCreate"));
 			return admin;
 		}
 
@@ -338,6 +343,11 @@ public NewTopics topics456() {
 						.build());
 		}
 
+		@Bean
+		NewTopic dontCreateThisOne() {
+			return TopicBuilder.name("dontCreate").build();
+		}
+
 	}
 
 }