GH-1538: Embedded broker get results for addTopics

Resolves #1538

**cherry-pick to 2.5.x**

# Conflicts:
#	spring-kafka/src/test/java/org/springframework/kafka/listener/ABSwitchClusterTests.java

Add docs.

spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java

@@ -22,6 +22,7 @@
 import java.net.InetSocketAddress;
 import java.nio.file.Files;
 import java.time.Duration;
+import java.util.AbstractMap.SimpleEntry;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -33,8 +34,11 @@
 import java.util.Properties;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import org.apache.commons.logging.LogFactory;
@@ -417,22 +421,100 @@ private void createTopics(AdminClient admin, List<NewTopic> newTopics) {
 		}
 	}
 
+	/**
+	 * Add topics to the existing broker(s) using the configured number of partitions.
+	 * The broker(s) must be running.
+	 * @param topicsToAdd the topics.
+	 * @return the results; null values indicate success.
+	 * @since 2.5.4
+	 */
+	public Map<String, Exception> addTopicsWithResults(String... topicsToAdd) {
+		Assert.notNull(this.zookeeper, "Broker must be started before this method can be called");
+		HashSet<String> set = new HashSet<>(Arrays.asList(topicsToAdd));
+		this.topics.addAll(set);
+		return createKafkaTopicsWithResults(set);
+	}
+
+	/**
+	 * Add topics to the existing broker(s) and returning a map of results.
+	 * The broker(s) must be running.
+	 * @param topicsToAdd the topics.
+	 * @return the results; null values indicate success.
+	 * @since 2.5.4
+	 */
+	public Map<String, Exception> addTopicsWithResults(NewTopic... topicsToAdd) {
+		Assert.notNull(this.zookeeper, "Broker must be started before this method can be called");
+		for (NewTopic topic : topicsToAdd) {
+			Assert.isTrue(this.topics.add(topic.name()), () -> "topic already exists: " + topic);
+			Assert.isTrue(topic.replicationFactor() <= this.count
+							&& (topic.replicasAssignments() == null
+							|| topic.replicasAssignments().size() <= this.count),
+					() -> "Embedded kafka does not support the requested replication factor: " + topic);
+		}
+
+		return doWithAdminFunction(admin -> createTopicsWithResults(admin, Arrays.asList(topicsToAdd)));
+	}
+
+	/**
+	 * Create topics in the existing broker(s) using the configured number of partitions
+	 * and returning a map of results.
+	 * @param topicsToCreate the topics.
+	 * @return the results; null values indicate success.
+	 * @since 2.5.4
+	 */
+	private Map<String, Exception> createKafkaTopicsWithResults(Set<String> topicsToCreate) {
+		return doWithAdminFunction(admin -> {
+			return createTopicsWithResults(admin,
+					topicsToCreate.stream()
+						.map(t -> new NewTopic(t, this.partitionsPerTopic, (short) this.count))
+						.collect(Collectors.toList()));
+		});
+	}
+
+	private Map<String, Exception> createTopicsWithResults(AdminClient admin, List<NewTopic> newTopics) {
+		CreateTopicsResult createTopics = admin.createTopics(newTopics);
+		Map<String, Exception> results = new HashMap<>();
+		createTopics.values()
+				.entrySet()
+				.stream()
+				.map(entry -> {
+					Exception result;
+					try {
+						entry.getValue().get(this.adminTimeout.getSeconds(), TimeUnit.SECONDS);
+						result = null;
+					}
+					catch (InterruptedException | ExecutionException | TimeoutException e) {
+						result = e;
+					}
+					return new SimpleEntry<>(entry.getKey(), result);
+				})
+				.forEach(entry -> results.put(entry.getKey(), entry.getValue()));
+		return results;
+	}
+
 	/**
 	 * Create an {@link AdminClient}; invoke the callback and reliably close the admin.
 	 * @param callback the callback.
 	 */
 	public void doWithAdmin(java.util.function.Consumer<AdminClient> callback) {
 		Map<String, Object> adminConfigs = new HashMap<>();
 		adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, getBrokersAsString());
-		AdminClient admin = null;
-		try {
-			admin = AdminClient.create(adminConfigs);
+		try (AdminClient admin = AdminClient.create(adminConfigs)) {
 			callback.accept(admin);
 		}
-		finally {
-			if (admin != null) {
-				admin.close(this.adminTimeout);
-			}
+	}
+
+	/**
+	 * Create an {@link AdminClient}; invoke the callback and reliably close the admin.
+	 * @param callback the callback.
+	 * @return a map of results.
+	 * @since 2.5.4
+	 */
+	public <T> T doWithAdminFunction(Function<AdminClient, T> callback) {
+		Map<String, Object> adminConfigs = new HashMap<>();
+		adminConfigs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, getBrokersAsString());
+		try (AdminClient admin = AdminClient.create(adminConfigs)) {
+			return callback.apply(admin);
 		}
 	}
 

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -39,6 +39,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -60,6 +61,7 @@
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.junit.jupiter.api.Test;
@@ -761,9 +763,22 @@ public void testConverterBean() throws Exception {
 	@Test
 	public void testAddingTopics() {
 		int count = this.embeddedKafka.getTopics().size();
-		this.embeddedKafka.addTopics("testAddingTopics");
+		Map<String, Exception> results = this.embeddedKafka.addTopicsWithResults("testAddingTopics");
+		assertThat(results).hasSize(1);
+		assertThat(results.keySet().iterator().next()).isEqualTo("testAddingTopics");
+		assertThat(results.get("testAddingTopics")).isNull();
 		assertThat(this.embeddedKafka.getTopics().size()).isEqualTo(count + 1);
-		this.embeddedKafka.addTopics(new NewTopic("morePartitions", 10, (short) 1));
+		results = this.embeddedKafka.addTopicsWithResults("testAddingTopics");
+		assertThat(results).hasSize(1);
+		assertThat(results.keySet().iterator().next()).isEqualTo("testAddingTopics");
+		assertThat(results.get("testAddingTopics"))
+				.isInstanceOf(ExecutionException.class)
+				.hasCauseInstanceOf(TopicExistsException.class);
+		assertThat(this.embeddedKafka.getTopics().size()).isEqualTo(count + 1);
+		results = this.embeddedKafka.addTopicsWithResults(new NewTopic("morePartitions", 10, (short) 1));
+		assertThat(results).hasSize(1);
+		assertThat(results.keySet().iterator().next()).isEqualTo("morePartitions");
+		assertThat(results.get("morePartitions")).isNull();
 		assertThat(this.embeddedKafka.getTopics().size()).isEqualTo(count + 2);
 		assertThatIllegalArgumentException()
 				.isThrownBy(() -> this.embeddedKafka.addTopics(new NewTopic("morePartitions", 10, (short) 1)))

src/reference/asciidoc/testing.adoc

@@ -142,6 +142,9 @@ public class MyTests {
 ----
 ====
 
+By default, `addTopics` will throw an exception when problems arise (such as adding a topic that already exists).
+Version 2.6 added a new version of that method that returns a `Map<String, Exception>`; the key is the topic name and the value is `null` for success, or an `Exception` for a failure.
+
 ==== Using the Same Brokers for Multiple Test Classes
 
 There is no built-in support for doing so, but you can use the same broker for multiple test classes with something similar to the following: