Set bootstrap servers property in @embeddedkafka

Author: garyrussell

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

@@ -200,6 +200,17 @@ public void setAdminTimeout(int adminTimeout) {
 		this.adminTimeout = Duration.ofSeconds(adminTimeout);
 	}
 
+	/**
+	 * Set the system property with this name to the list of broker addresses.
+	 * @param brokerListProperty the brokerListProperty to set
+	 * @return this broker.
+	 * @since 2.3
+	 */
+	public EmbeddedKafkaBroker brokerListProperty(String brokerListProperty) {
+		this.brokerListProperty = brokerListProperty;
+		return this;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		this.zookeeper = new EmbeddedZookeeper();
@@ -227,7 +238,9 @@ public void afterPropertiesSet() {
 			}
 		}
 		createKafkaTopics(this.topics);
-		this.brokerListProperty = System.getProperty(BROKER_LIST_PROPERTY);
+		if (this.brokerListProperty == null) {
+			this.brokerListProperty = System.getProperty(BROKER_LIST_PROPERTY);
+		}
 		if (this.brokerListProperty == null) {
 			this.brokerListProperty = SPRING_EMBEDDED_KAFKA_BROKERS;
 		}

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

@@ -151,6 +151,9 @@ private EmbeddedKafkaBroker createBroker(EmbeddedKafka embedded) {
 			}
 		}
 		broker.brokerProperties((Map<String, String>) (Map<?, ?>) properties);
+		if (StringUtils.hasText(embedded.bootstrapServersProperty())) {
+			broker.brokerListProperty(embedded.bootstrapServersProperty());
+		}
 		broker.afterPropertiesSet();
 		return broker;
 	}

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

@@ -133,5 +133,14 @@
 	 */
 	String brokerPropertiesLocation() default "";
 
+	/**
+	 * The property name to set with the bootstrap server addresses instead of the default
+	 * {@value org.springframework.kafka.test.EmbeddedKafkaBroker#SPRING_EMBEDDED_KAFKA_BROKERS}.
+	 * @return the property name.
+	 * @since 2.3
+	 * @see org.springframework.kafka.test.EmbeddedKafkaBroker#brokerListProperty(String)
+	 */
+	String bootstrapServersProperty() default "";
+
 }
 

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

@@ -105,6 +105,9 @@ public void customizeContext(ConfigurableApplicationContext context, MergedConte
 		}
 
 		embeddedKafkaBroker.brokerProperties((Map<String, String>) (Map<?, ?>) properties);
+		if (StringUtils.hasText(this.embeddedKafka.bootstrapServersProperty())) {
+			embeddedKafkaBroker.brokerListProperty(this.embeddedKafka.bootstrapServersProperty());
+		}
 
 		beanFactory.initializeBean(embeddedKafkaBroker, EmbeddedKafkaBroker.BEAN_NAME);
 		beanFactory.registerSingleton(EmbeddedKafkaBroker.BEAN_NAME, embeddedKafkaBroker);

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

@@ -22,18 +22,20 @@
 
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 /**
  * @author Gary Russell
  * @since 2.3
  *
  */
-@EmbeddedKafka
+@EmbeddedKafka(bootstrapServersProperty = "my.bss.property")
 public class EmbeddedKafkaConditionTests {
 
 	@Test
 	public void test(EmbeddedKafkaBroker broker) {
 		assertThat(broker.getBrokersAsString()).isNotNull();
+		assertThat(KafkaTestUtils.getPropertyValue(broker, "brokerListProperty")).isEqualTo("my.bss.property");
 	}
 
 }

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

@@ -18,7 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.BDDMockito.given;
-import static org.mockito.BDDMockito.mock;
+import static org.mockito.Mockito.mock;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -30,12 +30,14 @@
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 
 /**
  * @author Oleg Artyomov
  * @author Sergio Lourenco
  * @author Artem Bilan
+ * @author Gary Russell
  *
  * @since 1.3
  */
@@ -81,6 +83,8 @@ public void testPorts() {
 
 		assertThat(factoryStub.getBroker().getBrokersAsString())
 				.isEqualTo("127.0.0.1:" + annotationWithPorts.ports()[0]);
+		assertThat(KafkaTestUtils.getPropertyValue(factoryStub.getBroker(), "brokerListProperty"))
+				.isEqualTo("my.bss.prop");
 	}
 
 
@@ -94,7 +98,7 @@ private class SecondTestWithEmbeddedKafka {
 
 	}
 
-	@EmbeddedKafka(ports = 8085)
+	@EmbeddedKafka(ports = 8085, bootstrapServersProperty = "my.bss.prop")
 	private class TestWithEmbeddedKafkaPorts {
 
 	}

src/reference/asciidoc/testing.adoc

@@ -338,12 +338,8 @@ public class MyApplicationTests {
 
     @ClassRule
     public static EmbeddedKafkaRule broker = new EmbeddedKafkaRule(1,
-        false, "someTopic");
-
-    @BeforeClass
-    public static void setup() {
-        System.setProperty("spring.kafka.bootstrap-servers",
-                    broker.getEmbeddedKafka().getBrokersAsString());
+        false, "someTopic")
+            .brokerListProperty("spring.kafka.bootstrap-servers");
     }
 
     @Autowired
@@ -358,6 +354,8 @@ public class MyApplicationTests {
 ----
 ====
 
+Notice that, since this is a Spring Boot application, we override the broker list property to set Boot's property.
+
 [[kafka-testing-embeddedkafka-annotation]]
 ===== `@EmbeddedKafka` Annotation or `EmbeddedKafkaBroker` Bean
 
@@ -367,14 +365,10 @@ The following example shows how to use an `@EmbeddedKafka` Annotation to create
 [source, java]
 ----
 @RunWith(SpringRunner.class)
-@EmbeddedKafka(topics = "someTopic")
+@EmbeddedKafka(topics = "someTopic",
+        bootstrapServersProperty = "spring.kafka.bootstrap-servers")
 public class MyApplicationTests {
 
-    static {
-        System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY,
-                "spring.kafka.bootstrap-servers");
-    }
-
     @Autowired
     private KafkaTemplate<String, String> template;
 

src/reference/asciidoc/whats-new.adoc

@@ -124,3 +124,8 @@ See <<retrying-deserialization>> for more information.
 
 `ErrorHandlingDeserializer2` now uses a POJO (`FailedDeserializationInfo`) for passing all the contextual information around a deserialization error.
 This enables the code to access to extra information that was missing in the old `BiFunction<byte[], Headers, T> failedDeserializationFunction`.
+
+==== EmbeddedKafkaBroker Changes
+
+You can now override the default broker list property name in the annotation.
+See <<kafka-testing-embeddedkafka-annotation>> for more information.