> for more information.
\ No newline at end of file
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
new file mode 100644
index 00000000..0edec15a
--- /dev/null
+++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListener.java
@@ -0,0 +1,145 @@
+/*
+ * Copyright 2022 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.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.test.junit;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.platform.engine.ConfigurationParameters;
+import org.junit.platform.launcher.TestExecutionListener;
+import org.junit.platform.launcher.TestPlan;
+
+import org.springframework.core.io.DefaultResourceLoader;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PropertiesLoaderUtils;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.util.StringUtils;
+
+/**
+ * The {@link TestExecutionListener} to start an {@link EmbeddedKafkaBroker}
+ * in the beginning of the test plan and stop in the end.
+ * This approach ensures one global Kafka cluster for all the unit tests to execute.
+ *
+ * The {@link GlobalEmbeddedKafkaTestExecutionListener} is disabled by default.
+ * Set {@link GlobalEmbeddedKafkaTestExecutionListener#LISTENER_ENABLED_PROPERTY_NAME}
+ * system property (or respective {@link ConfigurationParameters#CONFIG_FILE_NAME} entry)
+ * to enable it.
+ *
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+public class GlobalEmbeddedKafkaTestExecutionListener implements TestExecutionListener {
+
+ private static final Log LOGGER = LogFactory.getLog(GlobalEmbeddedKafkaTestExecutionListener.class);
+
+ /**
+ * Property name used to enable the {@code GlobalEmbeddedKafkaTestExecutionListener}.
+ * The {@code GlobalEmbeddedKafkaTestExecutionListener} is registered automatically via
+ * Java's {@link java.util.ServiceLoader} mechanism but disabled by default.
+ * Set the value of this property to {@code true} to enable this listener.
+ */
+ public static final String LISTENER_ENABLED_PROPERTY_NAME = "spring.kafka.global.embedded.enabled";
+
+ /**
+ * The number of brokers for {@link EmbeddedKafkaBroker}.
+ */
+ public static final String COUNT_PROPERTY_NAME = "spring.kafka.embedded.count";
+
+ /**
+ * The port(s) to expose embedded broker(s).
+ */
+ public static final String PORTS_PROPERTY_NAME = "spring.kafka.embedded.ports";
+
+ /**
+ * The topics to create on the embedded broker(s).
+ */
+ public static final String TOPICS_PROPERTY_NAME = "spring.kafka.embedded.topics";
+
+ /**
+ * The number of partitions on topics to create on the embedded broker(s).
+ */
+ public static final String PARTITIONS_PROPERTY_NAME = "spring.kafka.embedded.partitions";
+
+ /**
+ * The location for a properties file with Kafka broker configuration.
+ */
+ public static final String BROKER_PROPERTIES_LOCATION_PROPERTY_NAME =
+ "spring.kafka.embedded.broker.properties.location";
+
+ private EmbeddedKafkaBroker embeddedKafkaBroker;
+
+ @Override
+ public void testPlanExecutionStarted(TestPlan testPlan) {
+ ConfigurationParameters configurationParameters = testPlan.getConfigurationParameters();
+ boolean enabled = configurationParameters.getBoolean(LISTENER_ENABLED_PROPERTY_NAME).orElse(false);
+ if (enabled) {
+ Integer count = configurationParameters.get(COUNT_PROPERTY_NAME, Integer::parseInt).orElse(1);
+ String[] topics =
+ configurationParameters.get(TOPICS_PROPERTY_NAME, StringUtils::commaDelimitedListToStringArray)
+ .orElse(null);
+ Integer partitions = configurationParameters.get(PARTITIONS_PROPERTY_NAME, Integer::parseInt).orElse(2);
+ Map brokerProperties =
+ configurationParameters.get(BROKER_PROPERTIES_LOCATION_PROPERTY_NAME, this::brokerProperties)
+ .orElse(null);
+ String brokerListProperty = configurationParameters.get(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY)
+ .orElse(null);
+ int[] ports =
+ configurationParameters.get(PORTS_PROPERTY_NAME, this::ports)
+ .orElse(new int[count]);
+
+ this.embeddedKafkaBroker =
+ new EmbeddedKafkaBroker(count, false, partitions, topics)
+ .brokerProperties(brokerProperties)
+ .brokerListProperty(brokerListProperty)
+ .kafkaPorts(ports);
+ this.embeddedKafkaBroker.afterPropertiesSet();
+
+ LOGGER.info("Started global Embedded Kafka on: " + this.embeddedKafkaBroker.getBrokersAsString());
+ }
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private Map brokerProperties(String propertiesLocation) {
+ Resource propertiesResource = new DefaultResourceLoader().getResource(propertiesLocation);
+ try {
+ return (Map) PropertiesLoaderUtils.loadProperties(propertiesResource);
+ }
+ catch (IOException ex) {
+ throw new UncheckedIOException(ex);
+ }
+ }
+
+ private int[] ports(String ports) {
+ return StringUtils.commaDelimitedListToSet(ports)
+ .stream()
+ .mapToInt(Integer::parseInt)
+ .toArray();
+ }
+
+ @Override
+ public void testPlanExecutionFinished(TestPlan testPlan) {
+ if (this.embeddedKafkaBroker != null) {
+ this.embeddedKafkaBroker.destroy();
+ LOGGER.info("Stopped global Embedded Kafka.");
+ }
+ }
+
+}
diff --git a/spring-kafka-test/src/main/resources/META-INF/services/org.junit.platform.launcher.TestExecutionListener b/spring-kafka-test/src/main/resources/META-INF/services/org.junit.platform.launcher.TestExecutionListener
new file mode 100644
index 00000000..28af8b91
--- /dev/null
+++ b/spring-kafka-test/src/main/resources/META-INF/services/org.junit.platform.launcher.TestExecutionListener
@@ -0,0 +1 @@
+org.springframework.kafka.test.junit.GlobalEmbeddedKafkaTestExecutionListener
diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/WithSpringTestContextTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/WithSpringTestContextTests.java
index 4a990b92..337af55f 100644
--- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/WithSpringTestContextTests.java
+++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/condition/WithSpringTestContextTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2021 the original author or authors.
+ * Copyright 2021-2022 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.
@@ -24,6 +24,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
+import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
@@ -33,6 +34,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
*/
@SpringJUnitConfig
@EmbeddedKafka
+@DirtiesContext
public class WithSpringTestContextTests {
@Test
diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListenerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListenerTests.java
new file mode 100644
index 00000000..f245e141
--- /dev/null
+++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/junit/GlobalEmbeddedKafkaTestExecutionListenerTests.java
@@ -0,0 +1,151 @@
+/*
+ * Copyright 2022 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.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.kafka.test.junit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.junit.platform.engine.discovery.DiscoverySelectors;
+import org.junit.platform.launcher.core.LauncherDiscoveryRequestBuilder;
+import org.junit.platform.launcher.core.LauncherFactory;
+import org.junit.platform.launcher.listeners.SummaryGeneratingListener;
+
+import org.springframework.util.DefaultPropertiesPersister;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 3.0
+ */
+public class GlobalEmbeddedKafkaTestExecutionListenerTests {
+
+ @BeforeAll
+ static void setup() {
+ System.setProperty(GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME, "true");
+ }
+
+ @AfterAll
+ static void tearDown() {
+ System.clearProperty(GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME);
+ System.clearProperty(GlobalEmbeddedKafkaTestExecutionListener.BROKER_PROPERTIES_LOCATION_PROPERTY_NAME);
+ }
+
+ @Test
+ void testGlobalEmbeddedKafkaTestExecutionListener() throws IOException {
+ var brokerProperties = new Properties();
+ brokerProperties.setProperty("auto.create.topics.enable", "false");
+
+ var propertiesFile = File.createTempFile("kafka-broker", ".properties");
+
+ try (var outputStream = new BufferedOutputStream(new FileOutputStream(propertiesFile))) {
+ new DefaultPropertiesPersister().store(brokerProperties, outputStream, "Last entry");
+ }
+
+ System.setProperty(GlobalEmbeddedKafkaTestExecutionListener.BROKER_PROPERTIES_LOCATION_PROPERTY_NAME,
+ "file:" + propertiesFile.getAbsolutePath());
+
+ var discoveryRequest =
+ LauncherDiscoveryRequestBuilder.request()
+ .selectors(DiscoverySelectors.selectClass(TestClass1.class),
+ DiscoverySelectors.selectClass(TestClass2.class))
+ .build();
+
+ var summaryGeneratingListener = new SummaryGeneratingListener();
+ LauncherFactory.create().execute(discoveryRequest, summaryGeneratingListener);
+
+ var summary = summaryGeneratingListener.getSummary();
+
+ try {
+ assertThat(summary.getTestsStartedCount()).isEqualTo(2);
+ assertThat(summary.getTestsSucceededCount()).isEqualTo(1);
+ assertThat(summary.getTestsFailedCount()).isEqualTo(1);
+ }
+ catch (Exception ex) {
+ summary.printFailuresTo(new PrintWriter(System.out));
+ throw ex;
+ }
+ }
+
+ @EnabledIfSystemProperty(named = GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME,
+ matches = "true")
+ static class TestClass1 {
+
+ @Test
+ void testDescribeTopic() throws ExecutionException, InterruptedException, TimeoutException {
+ Map adminConfigs =
+ Map.of(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
+ System.getProperty("spring.kafka.bootstrap-servers"));
+ try (var admin = AdminClient.create(adminConfigs)) {
+ var topicsMap =
+ admin.describeTopics(Set.of("topic1", "topic2"))
+ .allTopicNames()
+ .get(10, TimeUnit.SECONDS);
+
+ assertThat(topicsMap).containsOnlyKeys("topic1", "topic2");
+ }
+ }
+
+ }
+
+ @EnabledIfSystemProperty(named = GlobalEmbeddedKafkaTestExecutionListener.LISTENER_ENABLED_PROPERTY_NAME,
+ matches = "true")
+ static class TestClass2 {
+
+ @Test
+ void testCannotAutoCreateTopic() throws ExecutionException, InterruptedException, TimeoutException {
+ Map producerConfigs = new HashMap<>();
+ producerConfigs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
+ System.getProperty("spring.kafka.bootstrap-servers"));
+ producerConfigs.put(ProducerConfig.RETRIES_CONFIG, 1);
+ producerConfigs.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1);
+ producerConfigs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10);
+
+ StringSerializer serializer = new StringSerializer();
+ try (var kafkaProducer = new KafkaProducer<>(producerConfigs, serializer, serializer)) {
+ var recordMetadata =
+ kafkaProducer.send(new ProducerRecord<>("nonExistingTopic", "testValue"))
+ .get(10, TimeUnit.SECONDS);
+
+ assertThat(recordMetadata).isNotNull();
+ }
+ }
+
+ }
+
+}
diff --git a/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java b/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java
index 23c28497..6c4803bf 100644
--- a/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java
+++ b/spring-kafka-test/src/test/java/org/springframework/kafka/test/rule/AddressableEmbeddedBrokerTests.java
@@ -1,5 +1,5 @@
/*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -36,6 +36,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.utils.KafkaTestUtils;
+import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
@@ -48,6 +49,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
*
*/
@SpringJUnitConfig
+@DirtiesContext
public class AddressableEmbeddedBrokerTests {
private static final String TEST_EMBEDDED = "testEmbedded";
diff --git a/spring-kafka-test/src/test/resources/junit-platform.properties b/spring-kafka-test/src/test/resources/junit-platform.properties
new file mode 100644
index 00000000..0c8701ae
--- /dev/null
+++ b/spring-kafka-test/src/test/resources/junit-platform.properties
@@ -0,0 +1,3 @@
+spring.kafka.embedded.count=2
+spring.embedded.kafka.brokers.property=spring.kafka.bootstrap-servers
+spring.kafka.embedded.topics=topic1,topic2