GH-1347: Fix @embeddedkafka with count > 1

Resolves #1347

When count > 1 with no port specification (default random port)
tests failed because not enough ports were provided,

Author: garyrussell

spring-kafka-test/src/main/java/org/springframework/kafka/test/condition/EmbeddedKafkaCondition.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -115,9 +115,13 @@ private boolean springTestContext(AnnotatedElement annotatedElement) {
 	@SuppressWarnings("unchecked")
 	private EmbeddedKafkaBroker createBroker(EmbeddedKafka embedded) {
 		EmbeddedKafkaBroker broker;
+		int[] ports = embedded.ports();
+		if (embedded.count() > 1 && ports.length == 1 && ports[0] == 0) {
+			ports = new int[embedded.count()];
+		}
 		broker = new EmbeddedKafkaBroker(embedded.count(), embedded.controlledShutdown(), embedded.topics())
 				.zkPort(embedded.zookeeperPort())
-				.kafkaPorts(embedded.ports());
+				.kafkaPorts(ports);
 		Properties properties = new Properties();
 
 		for (String pair : embedded.brokerProperties()) {

spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -93,7 +93,7 @@
 	 * @return ports for brokers.
 	 * @since 2.2.4
 	 */
-	int[] ports() default {0};
+	int[] ports() default { 0 };
 
 	/**
 	 * Set the port on which the embedded Zookeeper should listen;

spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -66,11 +66,15 @@ public void customizeContext(ConfigurableApplicationContext context, MergedConte
 						.map(environment::resolvePlaceholders)
 						.toArray(String[]::new);
 
+		int[] ports = this.embeddedKafka.ports();
+		if (this.embeddedKafka.count() > 1 && ports.length == 1 && ports[0] == 0) {
+			ports = new int[this.embeddedKafka.count()];
+		}
 		EmbeddedKafkaBroker embeddedKafkaBroker = new EmbeddedKafkaBroker(this.embeddedKafka.count(),
 					this.embeddedKafka.controlledShutdown(),
 					this.embeddedKafka.partitions(),
 					topics)
-				.kafkaPorts(this.embeddedKafka.ports())
+				.kafkaPorts(ports)
 				.zkPort(this.embeddedKafka.zookeeperPort());
 
 		Properties properties = new Properties();

spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/EmbeddedKafkaConditionTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2020 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.
@@ -29,7 +29,7 @@
  * @since 2.3
  *
  */
-@EmbeddedKafka(bootstrapServersProperty = "my.bss.property")
+@EmbeddedKafka(bootstrapServersProperty = "my.bss.property", count = 2)
 public class EmbeddedKafkaConditionTests {
 
 	@Test

spring-kafka-test/src/test/java/org/springframework/kafka/test/context/EmbeddedKafkaContextCustomizerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2020 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.
@@ -48,30 +48,30 @@ public class EmbeddedKafkaContextCustomizerTests {
 	private EmbeddedKafka annotationFromSecondClass;
 
 	@BeforeEach
-	public void beforeEachTest() {
+	void beforeEachTest() {
 		annotationFromFirstClass = AnnotationUtils.findAnnotation(TestWithEmbeddedKafka.class, EmbeddedKafka.class);
 		annotationFromSecondClass =
 				AnnotationUtils.findAnnotation(SecondTestWithEmbeddedKafka.class, EmbeddedKafka.class);
 	}
 
 
 	@Test
-	public void testHashCode() {
+	void testHashCode() {
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode()).isNotEqualTo(0);
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass).hashCode())
 				.isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass).hashCode());
 	}
 
 
 	@Test
-	public void testEquals() {
+	void testEquals() {
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass))
 				.isEqualTo(new EmbeddedKafkaContextCustomizer(annotationFromSecondClass));
 		assertThat(new EmbeddedKafkaContextCustomizer(annotationFromFirstClass)).isNotEqualTo(new Object());
 	}
 
 	@Test
-	public void testPorts() {
+	void testPorts() {
 		EmbeddedKafka annotationWithPorts =
 				AnnotationUtils.findAnnotation(TestWithEmbeddedKafkaPorts.class, EmbeddedKafka.class);
 		EmbeddedKafkaContextCustomizer customizer = new EmbeddedKafkaContextCustomizer(annotationWithPorts);
@@ -87,6 +87,21 @@ public void testPorts() {
 				.isEqualTo("my.bss.prop");
 	}
 
+	@Test
+	void testMulti() {
+		EmbeddedKafka annotationWithPorts =
+				AnnotationUtils.findAnnotation(TestWithEmbeddedKafkaMulti.class, EmbeddedKafka.class);
+		EmbeddedKafkaContextCustomizer customizer = new EmbeddedKafkaContextCustomizer(annotationWithPorts);
+		ConfigurableApplicationContext context = mock(ConfigurableApplicationContext.class);
+		BeanFactoryStub factoryStub = new BeanFactoryStub();
+		given(context.getBeanFactory()).willReturn(factoryStub);
+		given(context.getEnvironment()).willReturn(mock(ConfigurableEnvironment.class));
+		customizer.customizeContext(context, null);
+
+		assertThat(factoryStub.getBroker().getBrokersAsString())
+			.matches("127.0.0.1:[0-9]+,127.0.0.1:[0-9]+");
+	}
+
 
 	@EmbeddedKafka
 	private class TestWithEmbeddedKafka {
@@ -103,6 +118,10 @@ private class TestWithEmbeddedKafkaPorts {
 
 	}
 
+	@EmbeddedKafka(count = 2)
+	private class TestWithEmbeddedKafkaMulti {
+
+	}
 	@SuppressWarnings("serial")
 	private class BeanFactoryStub extends DefaultListableBeanFactory {