GH-676: Allow to add topics into KafkaEmbedded

Fixes #676

Author: nklmish

spring-kafka-test/src/main/java/org/springframework/kafka/test/rule/KafkaEmbedded.java

@@ -26,9 +26,11 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -76,6 +78,7 @@
  * @author Gary Russell
  * @author Kamill Sokol
  * @author Elliot Kennedy
+ * @author Nakul Mishra
  */
 public class KafkaEmbedded extends ExternalResource implements KafkaRule, InitializingBean, DisposableBean {
 
@@ -118,7 +121,7 @@ public class KafkaEmbedded extends ExternalResource implements KafkaRule, Initia
 
 	private final boolean controlledShutdown;
 
-	private final String[] topics;
+	private final Set<String> topics;
 
 	private final int partitionsPerTopic;
 
@@ -160,10 +163,10 @@ public KafkaEmbedded(int count, boolean controlledShutdown, int partitions, Stri
 		this.kafkaPorts = new int[this.count]; // random ports by default.
 		this.controlledShutdown = controlledShutdown;
 		if (topics != null) {
-			this.topics = topics;
+			this.topics = new HashSet<>(Arrays.asList(topics));
 		}
 		else {
-			this.topics = new String[0];
+			this.topics = new HashSet<>();
 		}
 		this.partitionsPerTopic = partitions;
 	}
@@ -235,19 +238,18 @@ public void before() throws Exception { //NOSONAR
 				this.kafkaPorts[i] = TestUtils.boundPort(server, SecurityProtocol.PLAINTEXT);
 			}
 		}
-		addTopics(this.topics);
+		createKafkaTopics(this.topics);
 		System.setProperty(SPRING_EMBEDDED_KAFKA_BROKERS, getBrokersAsString());
 		System.setProperty(SPRING_EMBEDDED_ZOOKEEPER_CONNECT, getZookeeperConnectionString());
 	}
 
 	/**
-	 * Add topics to the existing broker(s) using the configured number of partitions.
+	 * Create topics in the existing broker(s) using the configured number of partitions.
 	 * @param topics the topics.
-	 * @since 2.1
 	 */
-	public void addTopics(String... topics) {
+	private void createKafkaTopics(Set<String> topics) {
 		doWithAdmin(admin -> {
-			List<NewTopic> newTopics = Arrays.stream(topics)
+			List<NewTopic> newTopics = topics.stream()
 					.map(t -> new NewTopic(t, this.partitionsPerTopic, (short) this.count))
 					.collect(Collectors.toList());
 			CreateTopicsResult createTopics = admin.createTopics(newTopics);
@@ -260,6 +262,18 @@ public void addTopics(String... topics) {
 		});
 	}
 
+
+	/**
+	 * Add topics to the existing broker(s) using the configured number of partitions.
+	 * @param topics the topics.
+	 * @since 2.1
+	 */
+	public void addTopics(String... topics) {
+		HashSet<String> set = new HashSet<>(Arrays.asList(topics));
+		createKafkaTopics(set);
+		this.topics.addAll(set);
+	}
+
 	/**
 	 * Create an {@link AdminClient} invoke the callback and reliable close the
 	 * admin.
@@ -340,6 +354,10 @@ public void after() {
 		}
 	}
 
+	public Set<String> getTopics() {
+		return new HashSet<>(this.topics);
+	}
+
 	@Override
 	public List<KafkaServer> getKafkaServers() {
 		return this.kafkaServers;
@@ -457,7 +475,7 @@ public boolean isEmbedded() {
 	 * @throws Exception an exception.
 	 */
 	public void consumeFromAllEmbeddedTopics(Consumer<?, ?> consumer) throws Exception {
-		consumeFromEmbeddedTopics(consumer, this.topics);
+		consumeFromEmbeddedTopics(consumer, this.topics.toArray(new String[0]));
 	}
 
 	/**
@@ -477,9 +495,11 @@ public void consumeFromAnEmbeddedTopic(Consumer<?, ?> consumer, String topic) th
 	 * @throws Exception an exception.
 	 */
 	public void consumeFromEmbeddedTopics(Consumer<?, ?> consumer, String... topics) throws Exception {
-		for (String topic : topics) {
-			assertThat(this.topics).as("topic '" + topic + "' is not in embedded topic list").contains(topic);
-		}
+		HashSet<String> diff = new HashSet<>(Arrays.asList(topics));
+		diff.removeAll(new HashSet<>(this.topics));
+		assertThat(this.topics)
+				.as("topic(s):'" + diff + "' are not in embedded topic list")
+				.containsAll(new HashSet<>(Arrays.asList(topics)));
 		final CountDownLatch consumerLatch = new CountDownLatch(1);
 		consumer.subscribe(Arrays.asList(topics), new ConsumerRebalanceListener() {
 

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

@@ -110,6 +110,7 @@
  * @author Dariusz Szablinski
  * @author Venil Noronha
  * @author Dimitri Penner
+ * @author Nakul Mishra
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -642,6 +643,13 @@ public void testConverterBean() throws Exception {
 		assertThat(this.config.listen16Message).isEqualTo("foobar");
 	}
 
+	@Test
+	public void testAddingTopics() {
+		int count = embeddedKafka.getTopics().size();
+		embeddedKafka.addTopics("testAddingTopics");
+		assertThat(embeddedKafka.getTopics().size()).isEqualTo(count + 1);
+	}
+
 	@Configuration
 	@EnableKafka
 	@EnableTransactionManagement(proxyTargetClass = true)