From 38abddfb703e6b1cb79bed9f0f6ced3f6618ffe6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 24 Jul 2023 11:24:47 -0400 Subject: [PATCH] GH-2750: EmbeddedKafka Set Boot's Bootstrap Prop. Resolves https://github.com/spring-projects/spring-kafka/issues/2750 Previously, setting the `bootstrapServersProperty` on `@EmbeddedKafka` (or `brokerListProperty` on the broker itself) caused the broker to set that property instead of the default property `spring.embedded.kafka.brokers`. Now, the custom property is set in addition to the default. Also, the default property name is `spring.kafka.bootstrap-servers`. This allows easier testing of Spring Boot applications, without the need for additional configuration. --- spring-kafka-docs/src/main/asciidoc/testing.adoc | 7 +++++-- .../src/main/asciidoc/whats-new.adoc | 2 ++ .../kafka/test/EmbeddedKafkaBroker.java | 15 +++++++++------ .../kafka/test/context/EmbeddedKafka.java | 7 ++++--- .../GlobalEmbeddedKafkaTestExecutionListener.java | 4 +++- .../kafka/test/EmbeddedKafkaBrokerTests.java | 8 +++++++- .../src/test/resources/junit-platform.properties | 1 - 7 files changed, 30 insertions(+), 14 deletions(-) diff --git a/spring-kafka-docs/src/main/asciidoc/testing.adoc b/spring-kafka-docs/src/main/asciidoc/testing.adoc index 9c57b277..85b6396c 100644 --- a/spring-kafka-docs/src/main/asciidoc/testing.adoc +++ b/spring-kafka-docs/src/main/asciidoc/testing.adoc @@ -133,6 +133,7 @@ Instead of default `spring.embedded.kafka.brokers` system property, the address For this purpose a `spring.embedded.kafka.brokers.property` (`EmbeddedKafkaBroker.BROKER_LIST_PROPERTY`) system property can be set before starting an embedded Kafka. For example, with Spring Boot a `spring.kafka.bootstrap-servers` configuration property is expected to be set for auto-configuring Kafka client, respectively. So, before running tests with an embedded Kafka on random ports, we can set `spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers` as a system property - and the `EmbeddedKafkaBroker` will use it to expose its broker addresses. +This is now the default value for this property (starting with version 3.0.10). With the `EmbeddedKafkaBroker.brokerProperties(Map)`, you can provide additional properties for the Kafka servers. See https://kafka.apache.org/documentation/#brokerconfigs[Kafka Config] for more information about possible broker properties. @@ -235,7 +236,8 @@ In addition, these properties can be provided: Essentially these properties mimic some of the `@EmbeddedKafka` attributes. See more information about configuration properties and how to provide them in the https://junit.org/junit5/docs/current/user-guide/#running-tests-config-params[JUnit 5 User Guide]. -For example, a `spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers` entry (for testing in Spring Boot application) can be added into a `junit-platform.properties` file in the testing classpath. +For example, a `spring.embedded.kafka.brokers.property=my.bootstrap-servers` entry can be added into a `junit-platform.properties` file in the testing classpath. +Starting with version 3.0.10, the broker automatically sets this to `spring.kafka.bootstrap-servers`, by default, for testing with Spring Boot applications. NOTE: It is recommended to not combine a global embedded Kafka and per-test class in a single test suite. Both of them share the same system properties, so it is very likely going to lead to unexpected behavior. @@ -422,7 +424,7 @@ The following example shows how to use an `@EmbeddedKafka` Annotation to create ---- @RunWith(SpringRunner.class) @EmbeddedKafka(topics = "someTopic", - bootstrapServersProperty = "spring.kafka.bootstrap-servers") + bootstrapServersProperty = "spring.kafka.bootstrap-servers") // this is now the default public class MyApplicationTests { @Autowired @@ -437,6 +439,7 @@ public class MyApplicationTests { ---- ==== +NOTE: The `bootstrapServersProperty` is automatically set to `spring.kafka.bootstrap-servers`, by default, starting with version 3.0.10. ==== Hamcrest Matchers diff --git a/spring-kafka-docs/src/main/asciidoc/whats-new.adoc b/spring-kafka-docs/src/main/asciidoc/whats-new.adoc index 9f02b7b1..d2384735 100644 --- a/spring-kafka-docs/src/main/asciidoc/whats-new.adoc +++ b/spring-kafka-docs/src/main/asciidoc/whats-new.adoc @@ -104,3 +104,5 @@ Similarly, `RECEIVED_MESSAGE_KEY` is replaced by `RECEIVED_KEY` and `RECEIVED_PA Version 3.0.7 introduced a `MockConsumerFactory` and `MockProducerFactory`. See <> for more information. + +Starting with version 3.0.10, the embedded Kafka broker, by default, sets the Spring Boot property `spring.kafka.bootstrap-servers` to the address(es) of the embedded broker(s). diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java index 71b309f8..6c32e7ce 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/EmbeddedKafkaBroker.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2022 the original author or authors. + * Copyright 2018-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. @@ -181,7 +181,7 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { private int zkSessionTimeout = DEFAULT_ZK_SESSION_TIMEOUT; - private String brokerListProperty; + private String brokerListProperty = "spring.kafka.bootstrap-servers"; private volatile ZooKeeperClient zooKeeperClient; @@ -257,6 +257,8 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { /** * Set the system property with this name to the list of broker addresses. + * Defaults to {@code spring.kafka.bootstrap-servers} for Spring Boot + * compatibility, since 3.0.10. * @param brokerListProperty the brokerListProperty to set * @return this broker. * @since 2.3 @@ -374,10 +376,10 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { if (this.brokerListProperty == null) { this.brokerListProperty = System.getProperty(BROKER_LIST_PROPERTY); } - if (this.brokerListProperty == null) { - this.brokerListProperty = SPRING_EMBEDDED_KAFKA_BROKERS; + if (this.brokerListProperty != null) { + System.setProperty(this.brokerListProperty, getBrokersAsString()); } - System.setProperty(this.brokerListProperty, getBrokersAsString()); + System.setProperty(SPRING_EMBEDDED_KAFKA_BROKERS, getBrokersAsString()); System.setProperty(SPRING_EMBEDDED_ZOOKEEPER_CONNECT, getZookeeperConnectionString()); } @@ -591,7 +593,8 @@ public class EmbeddedKafkaBroker implements InitializingBean, DisposableBean { @Override public void destroy() { - System.getProperties().remove(brokerListProperty); + System.getProperties().remove(this.brokerListProperty); + System.getProperties().remove(SPRING_EMBEDDED_KAFKA_BROKERS); System.getProperties().remove(SPRING_EMBEDDED_ZOOKEEPER_CONNECT); for (KafkaServer kafkaServer : this.kafkaServers) { try { diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java index 3b57204b..ef239f1c 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.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. @@ -151,13 +151,14 @@ public @interface EmbeddedKafka { String brokerPropertiesLocation() default ""; /** - * The property name to set with the bootstrap server addresses instead of the default + * The property name to set with the bootstrap server addresses as well as the default * {@value org.springframework.kafka.test.EmbeddedKafkaBroker#SPRING_EMBEDDED_KAFKA_BROKERS}. + * Defaults to {@code spring.kafka.bootstrap-servers} since 3.0.10. * @return the property name. * @since 2.3 * @see org.springframework.kafka.test.EmbeddedKafkaBroker#brokerListProperty(String) */ - String bootstrapServersProperty() default ""; + String bootstrapServersProperty() default "spring.kafka.bootstrap-servers"; /** * Timeout for internal ZK client connection. diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java index 8bb6fdd2..1c5956f4 100644 --- a/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java @@ -119,8 +119,10 @@ public class GlobalEmbeddedKafkaTestExecutionListener implements TestExecutionLi this.embeddedKafkaBroker = new EmbeddedKafkaBroker(count, false, partitions, topics) .brokerProperties(brokerProperties) - .brokerListProperty(brokerListProperty) .kafkaPorts(ports); + if (brokerListProperty != null) { + this.embeddedKafkaBroker.brokerListProperty(brokerListProperty); + } this.embeddedKafkaBroker.afterPropertiesSet(); this.logger.info("Started global Embedded Kafka on: " + this.embeddedKafkaBroker.getBrokersAsString()); diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java index 2d97bf26..458f4e42 100644 --- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java +++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaBrokerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-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. @@ -30,10 +30,16 @@ public class EmbeddedKafkaBrokerTests { @Test void testUpDown() { EmbeddedKafkaBroker kafka = new EmbeddedKafkaBroker(1); + kafka.brokerListProperty("foo.bar"); kafka.afterPropertiesSet(); assertThat(kafka.getZookeeperConnectionString()).startsWith("127"); + assertThat(System.getProperty("foo.bar")).isNotNull(); + assertThat(System.getProperty(EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS)) + .isEqualTo(System.getProperty("foo.bar")); kafka.destroy(); assertThat(kafka.getZookeeperConnectionString()).isNull(); + assertThat(System.getProperty("foo.bar")).isNull(); + assertThat(System.getProperty(EmbeddedKafkaBroker.SPRING_EMBEDDED_KAFKA_BROKERS)).isNull(); } } diff --git a/spring-kafka-test/src/test/resources/junit-platform.properties b/spring-kafka-test/src/test/resources/junit-platform.properties index 0c8701ae..603c9ee2 100644 --- a/spring-kafka-test/src/test/resources/junit-platform.properties +++ b/spring-kafka-test/src/test/resources/junit-platform.properties @@ -1,3 +1,2 @@ spring.kafka.embedded.count=2 -spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers spring.kafka.embedded.topics=topic1,topic2