Use Global Embedded Kafka whenever possible

For better test suite lifecycle (higher performance)
reuse one global embedded Kafka broker introduced in Spring for Apache Kafka `3.0`

Some tests have left with their own `EmbeddedKafkaBroker` definitions since they
rely on different partitions

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java

@@ -27,6 +27,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.channel.NullChannel;
@@ -44,8 +45,6 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ConsumerProperties;
 import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
@@ -61,7 +60,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  *
  */
 @SpringJUnitConfig
-@EmbeddedKafka(topics = { "channel.1", "channel.2", "channel.3" }, partitions = 1)
 public class ChannelTests {
 
 	@Test
@@ -118,17 +116,18 @@ public class ChannelTests {
 	@Configuration
 	public static class Config {
 
-		@Autowired
-		private EmbeddedKafkaBroker broker;
+		@Value("${spring.global.embedded.kafka.brokers}")
+		String embeddedKafkaBrokers;
 
 		@Bean
 		public ProducerFactory<Integer, String> pf() {
-			return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(this.broker));
+			return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(this.embeddedKafkaBrokers));
 		}
 
 		@Bean
 		public ConsumerFactory<Integer, String> cf() {
-			Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("channelTests", "false", this.broker);
+			Map<String, Object> consumerProps =
+					KafkaTestUtils.consumerProps(this.embeddedKafkaBrokers, "channelTests", "false");
 			consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 			return new DefaultKafkaConsumerFactory<>(consumerProps);
 		}

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml

@@ -3,14 +3,18 @@
 	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	   xmlns:int="http://www.springframework.org/schema/integration"
 	   xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
+	   xmlns:context="http://www.springframework.org/schema/context"
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
 		http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd
-		http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd">
+		http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd">
+
+	<context:property-placeholder />
 
 	<bean id="cf" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory">
 		<constructor-arg>
 			<map>
-				<entry key="bootstrap.servers" value="#{embeddedKafka.brokersAsString}"/>
+				<entry key="bootstrap.servers" value="${spring.global.embedded.kafka.brokers}"/>
 				<entry key="auto.offset.reset" value="earliest"/>
 				<entry key="key.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
 				<entry key="value.deserializer" value="org.apache.kafka.common.serialization.StringDeserializer"/>
@@ -22,7 +26,7 @@
 	<bean id="pf" class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
 		<constructor-arg>
 			<map>
-				<entry key="bootstrap.servers" value="#{embeddedKafka.brokersAsString}"/>
+				<entry key="bootstrap.servers" value="${spring.global.embedded.kafka.brokers}"/>
 				<entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
 				<entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer"/>
 			</map>

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java

@@ -22,7 +22,6 @@ import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.test.annotation.DirtiesContext;
@@ -36,7 +35,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
  */
 @SpringJUnitConfig
 @DirtiesContext
-@EmbeddedKafka(topics = { "one", "two", "three", "four" })
 public class AllXmlTests {
 
 	@Autowired

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020 the original author or authors.
+ * Copyright 2020-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.
@@ -28,16 +28,17 @@ import org.springframework.context.annotation.ImportResource;
 import org.springframework.integration.kafka.channel.PollableKafkaChannel;
 import org.springframework.integration.kafka.channel.SubscribableKafkaChannel;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.KafkaOperations;
 import org.springframework.kafka.listener.ConsumerProperties;
-import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  *
  * @since 5.4
  *
@@ -65,17 +66,17 @@ public class ChannelParserTests {
 
 	@Test
 	void testParser() {
-		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "topic")).isEqualTo("ptpTopic");
-		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "topic")).isEqualTo("pubSubTopic");
-		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "container")).isNotNull();
-		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "container")).isNotNull();
-		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "template")).isSameAs(this.template);
-		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "template")).isSameAs(this.template);
-		assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "template")).isSameAs(this.template);
-		assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "source")).isSameAs(this.source);
-		assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "groupId")).isEqualTo("ptpGroup");
-		assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "groupId")).isEqualTo("pubSubGroup");
-		assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "groupId")).isEqualTo("pollableGroup");
+		assertThat(TestUtils.getPropertyValue(this.ptp, "topic")).isEqualTo("ptpTopic");
+		assertThat(TestUtils.getPropertyValue(this.pubSub, "topic")).isEqualTo("pubSubTopic");
+		assertThat(TestUtils.getPropertyValue(this.ptp, "container")).isNotNull();
+		assertThat(TestUtils.getPropertyValue(this.pubSub, "container")).isNotNull();
+		assertThat(TestUtils.getPropertyValue(this.ptp, "template")).isSameAs(this.template);
+		assertThat(TestUtils.getPropertyValue(this.pubSub, "template")).isSameAs(this.template);
+		assertThat(TestUtils.getPropertyValue(this.pollable, "template")).isSameAs(this.template);
+		assertThat(TestUtils.getPropertyValue(this.pollable, "source")).isSameAs(this.source);
+		assertThat(TestUtils.getPropertyValue(this.ptp, "groupId")).isEqualTo("ptpGroup");
+		assertThat(TestUtils.getPropertyValue(this.pubSub, "groupId")).isEqualTo("pubSubGroup");
+		assertThat(TestUtils.getPropertyValue(this.pollable, "groupId")).isEqualTo("pollableGroup");
 	}
 
 	@Configuration

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java

@@ -35,6 +35,7 @@ import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -71,8 +72,6 @@ import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -97,9 +96,6 @@ import org.springframework.util.backoff.FixedBackOff;
  */
 @SpringJUnitConfig
 @DirtiesContext
-@EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3,
-		KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5, KafkaDslTests.TEST_TOPIC6, KafkaDslTests.TEST_TOPIC7,
-		KafkaDslTests.TEST_TOPIC8, KafkaDslTests.TEST_TOPIC9 })
 public class KafkaDslTests {
 
 	static final String TEST_TOPIC1 = "test-topic1";
@@ -118,8 +114,6 @@ public class KafkaDslTests {
 
 	static final String TEST_TOPIC8 = "test-topic8";
 
-	static final String TEST_TOPIC9 = "test-topic9";
-
 	@Autowired
 	@Qualifier("sendToKafkaFlow.input")
 	private MessageChannel sendToKafkaFlowInput;
@@ -266,7 +260,8 @@ public class KafkaDslTests {
 		assertThat(received)
 				.isNotNull()
 				.extracting("payload")
-				.isEqualTo("foo");	}
+				.isEqualTo("foo");
+	}
 
 	@Configuration
 	@EnableIntegration
@@ -281,14 +276,12 @@ public class KafkaDslTests {
 
 		private Object fromSource;
 
-		@Autowired
-		private EmbeddedKafkaBroker embeddedKafka;
-
+		@Value("${spring.global.embedded.kafka.brokers}")
+		String embeddedKafkaBrokers;
 
 		@Bean
 		public ConsumerFactory<Integer, String> consumerFactory() {
-			Map<String, Object> props = KafkaTestUtils
-					.consumerProps("test1", "false", this.embeddedKafka);
+			Map<String, Object> props = KafkaTestUtils.consumerProps(this.embeddedKafkaBrokers, "test1", "false");
 			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 			return new DefaultKafkaConsumerFactory<>(props);
 		}
@@ -302,7 +295,7 @@ public class KafkaDslTests {
 		public IntegrationFlow topic1ListenerFromKafkaFlow() {
 			return IntegrationFlow
 					.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
-							KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
+									KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
 							.configureListenerContainer(c ->
 									c.ackMode(ContainerProperties.AckMode.MANUAL)
 											.idleEventInterval(100L)
@@ -344,7 +337,7 @@ public class KafkaDslTests {
 
 		@Bean
 		public ProducerFactory<Integer, String> producerFactory() {
-			Map<String, Object> props = KafkaTestUtils.producerProps(this.embeddedKafka);
+			Map<String, Object> props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers);
 			props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
 			return new DefaultKafkaProducerFactory<>(props);
 		}
@@ -479,8 +472,8 @@ public class KafkaDslTests {
 		public IntegrationFlow serverGateway() {
 			return IntegrationFlow
 					.from(Kafka.inboundGateway(consumerFactory(), containerProperties(),
-							producerFactory())
-								.configureListenerContainer(container -> container.errorHandler(eh())))
+									producerFactory())
+							.configureListenerContainer(container -> container.errorHandler(eh())))
 					.<String, String>transform(String::toUpperCase)
 					.get();
 		}
@@ -495,9 +488,9 @@ public class KafkaDslTests {
 			ContainerProperties props = containerProperties();
 			props.setGroupId("wreh");
 			return IntegrationFlow.from(Kafka.messageDrivenChannelAdapter(consumerFactory(), props)
-						.configureListenerContainer(container -> {
-							container.errorHandler(recoveringErrorHandler());
-						}))
+							.configureListenerContainer(container -> {
+								container.errorHandler(recoveringErrorHandler());
+							}))
 					.handle(p -> {
 						throw new RuntimeException("test");
 					})

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java

@@ -50,6 +50,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -85,8 +86,6 @@ import org.springframework.kafka.support.converter.ConversionException;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.kafka.support.converter.StringJsonMessageConverter;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.ContainerTestUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
@@ -111,13 +110,6 @@ import org.springframework.retry.support.RetryTemplate;
  * @since 5.4
  *
  */
-@EmbeddedKafka(controlledShutdown = true,
-		topics = { MessageDrivenAdapterTests.topic1,
-				MessageDrivenAdapterTests.topic2,
-				MessageDrivenAdapterTests.topic3,
-				MessageDrivenAdapterTests.topic4,
-				MessageDrivenAdapterTests.topic5,
-				MessageDrivenAdapterTests.topic6 })
 class MessageDrivenAdapterTests {
 
 	static final String topic1 = "testTopic1";
@@ -132,9 +124,16 @@ class MessageDrivenAdapterTests {
 
 	static final String topic6 = "testTopic6";
 
+	static String EMBEDDED_BROKERS;
+
+	@BeforeAll
+	static void setup() {
+		EMBEDDED_BROKERS = System.getProperty("spring.global.embedded.kafka.brokers");
+	}
+
 	@Test
-	void testInboundRecord(EmbeddedKafkaBroker embeddedKafka) {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
+	void testInboundRecord() {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test1", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic1);
@@ -155,9 +154,9 @@ class MessageDrivenAdapterTests {
 
 		});
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic1);
@@ -219,8 +218,8 @@ class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	void testInboundRecordRetryRecover(EmbeddedKafkaBroker embeddedKafka) {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka);
+	void testInboundRecordRetryRecover() {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test4", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic4);
@@ -248,9 +247,9 @@ class MessageDrivenAdapterTests {
 		adapter.setRetryTemplate(retryTemplate);
 		adapter.afterPropertiesSet();
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic4);
@@ -286,8 +285,8 @@ class MessageDrivenAdapterTests {
 	 * to the consumer.
 	 */
 	@Test
-	void testInboundRecordRetryRecoverWithoutRecoveryCallback(EmbeddedKafkaBroker embeddedKafka) throws Exception {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka);
+	void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test6", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic6);
@@ -321,9 +320,9 @@ class MessageDrivenAdapterTests {
 
 		adapter.afterPropertiesSet();
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic6);
@@ -336,8 +335,8 @@ class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	void testInboundRecordNoRetryRecover(EmbeddedKafkaBroker embeddedKafka) {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka);
+	void testInboundRecordNoRetryRecover() {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test5", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic5);
@@ -362,9 +361,9 @@ class MessageDrivenAdapterTests {
 		adapter.setBindSourceRecord(true);
 		adapter.afterPropertiesSet();
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic5);
@@ -391,8 +390,8 @@ class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	void testInboundBatch(EmbeddedKafkaBroker embeddedKafka) throws Exception {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka);
+	void testInboundBatch() throws Exception {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test2", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic2);
@@ -421,9 +420,9 @@ class MessageDrivenAdapterTests {
 
 		});
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic2);
@@ -479,8 +478,8 @@ class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	void testInboundJson(EmbeddedKafkaBroker embeddedKafka) {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka);
+	void testInboundJson() {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test3", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic3);
@@ -492,9 +491,9 @@ class MessageDrivenAdapterTests {
 		adapter.setOutputChannel(out);
 		adapter.afterPropertiesSet();
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic3);
@@ -521,8 +520,8 @@ class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	void testInboundJsonWithPayload(EmbeddedKafkaBroker embeddedKafka) {
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka);
+	void testInboundJsonWithPayload() {
+		Map<String, Object> props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test6", "true");
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic6);
@@ -538,9 +537,9 @@ class MessageDrivenAdapterTests {
 		adapter.setOutputChannel(out);
 		adapter.afterPropertiesSet();
 		adapter.start();
-		ContainerTestUtils.waitForAssignment(container, 2);
+		ContainerTestUtils.waitForAssignment(container, 1);
 
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic6);

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java

@@ -34,8 +34,6 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.listener.ConsumerProperties;
-import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
@@ -48,14 +46,14 @@ import org.springframework.messaging.support.GenericMessage;
  * @since 5.4
  *
  */
-@EmbeddedKafka(controlledShutdown = true, topics = MessageSourceIntegrationTests.TOPIC1, partitions = 1)
 class MessageSourceIntegrationTests {
 
 	static final String TOPIC1 = "MessageSourceIntegrationTests1";
 
 	@Test
-	void testSource(EmbeddedKafkaBroker embeddedKafka) throws Exception {
-		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testSource", "false", embeddedKafka);
+	void testSource() throws Exception {
+		String brokers = System.getProperty("spring.global.embedded.kafka.brokers");
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(brokers, "testSource", "false");
 		consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		consumerProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 42);
@@ -81,7 +79,7 @@ class MessageSourceIntegrationTests {
 
 		KafkaMessageSource<Integer, String> source = new KafkaMessageSource<>(consumerFactory, consumerProperties);
 
-		Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
+		Map<String, Object> producerProps = KafkaTestUtils.producerProps(brokers);
 		DefaultKafkaProducerFactory<Object, Object> producerFactory = new DefaultKafkaProducerFactory<>(producerProps);
 		KafkaTemplate<Object, Object> template = new KafkaTemplate<>(producerFactory);
 		template.setDefaultTopic(TOPIC1);

spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt

@@ -23,10 +23,9 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.TopicPartition
 import org.junit.jupiter.api.Test
-import org.junit.jupiter.api.condition.DisabledOnOs
-import org.junit.jupiter.api.condition.OS
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.beans.factory.annotation.Value
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.integration.IntegrationMessageHeaderAccessor
@@ -48,8 +47,6 @@ import org.springframework.kafka.requestreply.ReplyingKafkaTemplate
 import org.springframework.kafka.support.Acknowledgment
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper
 import org.springframework.kafka.support.KafkaHeaders
-import org.springframework.kafka.test.EmbeddedKafkaBroker
-import org.springframework.kafka.test.context.EmbeddedKafka
 import org.springframework.kafka.test.utils.KafkaTestUtils
 import org.springframework.messaging.Message
 import org.springframework.messaging.MessageChannel
@@ -72,24 +69,21 @@ import java.util.stream.Stream
  * @since 5.4
  */
 
-@DisabledOnOs(OS.WINDOWS)
 @SpringJUnitConfig
 @DirtiesContext
-@EmbeddedKafka(topics = [KafkaDslKotlinTests.TEST_TOPIC1, KafkaDslKotlinTests.TEST_TOPIC2,
-	KafkaDslKotlinTests.TEST_TOPIC3, KafkaDslKotlinTests.TEST_TOPIC4, KafkaDslKotlinTests.TEST_TOPIC5])
 class KafkaDslKotlinTests {
 
 	companion object {
 
-		const val TEST_TOPIC1 = "test-topic1"
+		const val TEST_TOPIC1 = "test-kotlin-topic1"
 
-		const val TEST_TOPIC2 = "test-topic2"
+		const val TEST_TOPIC2 = "test-kotlin-topic2"
 
-		const val TEST_TOPIC3 = "test-topic3"
+		const val TEST_TOPIC3 = "test-kotlin-topic3"
 
-		const val TEST_TOPIC4 = "test-topic4"
+		const val TEST_TOPIC4 = "test-kotlin-topic4"
 
-		const val TEST_TOPIC5 = "test-topic5"
+		const val TEST_TOPIC5 = "test-kotlin-topic5"
 
 	}
 
@@ -108,10 +102,6 @@ class KafkaDslKotlinTests {
 	@Qualifier("kafkaProducer1.handler")
 	private lateinit var kafkaProducer1: KafkaProducerMessageHandler<*, *>
 
-	@Autowired
-	@Qualifier("kafkaProducer2.handler")
-	private lateinit var kafkaProducer2: KafkaProducerMessageHandler<*, *>
-
 	@Autowired
 	private lateinit var errorChannel: PollableChannel
 
@@ -120,11 +110,11 @@ class KafkaDslKotlinTests {
 	private lateinit var messageListenerContainer: MessageListenerContainer
 
 	@Autowired(required = false)
-	@Qualifier("kafkaTemplate:test-topic1")
+	@Qualifier("kafkaTemplate:test-kotlin-topic1")
 	private lateinit var kafkaTemplateTopic1: KafkaTemplate<Any, Any>
 
 	@Autowired(required = false)
-	@Qualifier("kafkaTemplate:test-topic2")
+	@Qualifier("kafkaTemplate:test-kotlin-topic2")
 	private lateinit var kafkaTemplateTopic2: KafkaTemplate<*, *>
 
 	@Autowired
@@ -215,12 +205,12 @@ class KafkaDslKotlinTests {
 
 		var fromSource: Any? = null
 
-		@Autowired
-		private lateinit var embeddedKafka: EmbeddedKafkaBroker
+		@Value("\${spring.global.embedded.kafka.brokers}")
+		lateinit var embeddedKafkaBrokers: String
 
 		@Bean
 		fun consumerFactory(): ConsumerFactory<Int, String> {
-			val props = KafkaTestUtils.consumerProps("test1", "false", this.embeddedKafka)
+			val props = KafkaTestUtils.consumerProps(this.embeddedKafkaBrokers, "test1", "false")
 			props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
 			return DefaultKafkaConsumerFactory(props)
 		}
@@ -230,96 +220,111 @@ class KafkaDslKotlinTests {
 
 		@Bean
 		fun topic1ListenerFromKafkaFlow() =
-				integrationFlow(
-						Kafka.messageDrivenChannelAdapter(consumerFactory(),
-								KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
-								.configureListenerContainer {
-									it.ackMode(ContainerProperties.AckMode.MANUAL)
-											.id("topic1ListenerContainer")
-								}
-								.errorChannel(errorChannel())
-								.retryTemplate(RetryTemplate())
-								.filterInRetry(true)) {
-					filter<Message<*>>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) { throwExceptionOnRejection(true) }
-					transform<String> { it.uppercase() }
-					channel { queue("listeningFromKafkaResults1") }
+			integrationFlow(
+				Kafka.messageDrivenChannelAdapter(
+					consumerFactory(),
+					KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1
+				)
+					.configureListenerContainer {
+						it.ackMode(ContainerProperties.AckMode.MANUAL)
+							.id("topic1ListenerContainer")
+					}
+					.errorChannel(errorChannel())
+					.retryTemplate(RetryTemplate())
+					.filterInRetry(true)) {
+				filter<Message<*>>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) {
+					throwExceptionOnRejection(
+						true
+					)
 				}
+				transform<String> { it.uppercase() }
+				channel { queue("listeningFromKafkaResults1") }
+			}
 
 		@Bean
 		fun topic2ListenerFromKafkaFlow() =
-				integrationFlow(
-						Kafka.messageDrivenChannelAdapter(consumerFactory(),
-								KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2)
-								.configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) }
-								.errorChannel(errorChannel())
-								.retryTemplate(RetryTemplate())
-								.filterInRetry(true)) {
-					filter<Message<*>>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) { throwExceptionOnRejection(true) }
-					transform<String> { it.uppercase() }
-					channel { queue("listeningFromKafkaResults2") }
+			integrationFlow(
+				Kafka.messageDrivenChannelAdapter(
+					consumerFactory(),
+					KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2
+				)
+					.configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) }
+					.errorChannel(errorChannel())
+					.retryTemplate(RetryTemplate())
+					.filterInRetry(true)) {
+				filter<Message<*>>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) {
+					throwExceptionOnRejection(
+						true
+					)
 				}
+				transform<String> { it.uppercase() }
+				channel { queue("listeningFromKafkaResults2") }
+			}
 
 		@Bean
 		fun producerFactory(): DefaultKafkaProducerFactory<Int, String> {
-			val props = KafkaTestUtils.producerProps(this.embeddedKafka)
+			val props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers)
 			props[ProducerConfig.MAX_BLOCK_MS_CONFIG] = "10000"
 			return DefaultKafkaProducerFactory(props)
 		}
 
 		@Bean
 		fun sendToKafkaFlow() =
-				integrationFlow {
-					split<String> { p -> Stream.generate { p }.limit(101) }
-					publishSubscribe(PublishSubscribeChannel(),
-							{
-								handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
-										.timestampExpression("T(Long).valueOf('1487694048633')")
-								) { id("kafkaProducer1") }
-							},
-							{
-								handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
-										.timestamp<Any> { 1487694048644L }
-								) { id("kafkaProducer2") }
-							}
-					)
-				}
+			integrationFlow {
+				split<String> { p -> Stream.generate { p }.limit(101) }
+				publishSubscribe(PublishSubscribeChannel(),
+					{
+						handle(
+							kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
+								.timestampExpression("T(Long).valueOf('1487694048633')")
+						) { id("kafkaProducer1") }
+					},
+					{
+						handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
+							.timestamp<Any> { 1487694048644L }
+						) { id("kafkaProducer2") }
+					}
+				)
+			}
 
 		@Bean
 		fun mapper() = DefaultKafkaHeaderMapper()
 
 		private fun kafkaMessageHandler(producerFactory: ProducerFactory<Int, String>, topic: String) =
-				Kafka.outboundChannelAdapter(producerFactory)
-						.messageKey<Any> { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
-						.headerMapper(mapper())
-						.sync(true)
-						.partitionId<Any> { 0 }
-						.topicExpression("headers[kafka_topic] ?: '$topic'")
-						.configureKafkaTemplate { it.id("kafkaTemplate:$topic") }
+			Kafka.outboundChannelAdapter(producerFactory)
+				.messageKey<Any> { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
+				.headerMapper(mapper())
+				.sync(true)
+				.partitionId<Any> { 0 }
+				.topicExpression("headers[kafka_topic] ?: '$topic'")
+				.configureKafkaTemplate { it.id("kafkaTemplate:$topic") }
 
 
 		@Bean
 		fun sourceFlow() =
-				integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)),
-						{ poller(Pollers.fixedDelay(100)) }) {
-					handle { m ->
-						this@ContextConfiguration.fromSource = m.payload
-						this@ContextConfiguration.sourceFlowLatch.countDown()
-					}
+			integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)),
+				{ poller(Pollers.fixedDelay(100)) }) {
+				handle { m ->
+					this@ContextConfiguration.fromSource = m.payload
+					this@ContextConfiguration.sourceFlowLatch.countDown()
 				}
+			}
 
 		@Bean
 		fun replyingKafkaTemplate() =
-				ReplyingKafkaTemplate(producerFactory(), replyContainer())
-						.also {
-							it.setDefaultReplyTimeout(Duration.ofSeconds(30))
-						}
+			ReplyingKafkaTemplate(producerFactory(), replyContainer())
+				.also {
+					it.setDefaultReplyTimeout(Duration.ofSeconds(30))
+				}
 
 		@Bean
 		fun outboundGateFlow() =
-				integrationFlow<Gate> {
-					handle(Kafka.outboundGateway(replyingKafkaTemplate())
-							.sync(true))
-				}
+			integrationFlow<Gate> {
+				handle(
+					Kafka.outboundGateway(replyingKafkaTemplate())
+						.sync(true)
+				)
+			}
 
 		private fun replyContainer(): GenericMessageListenerContainer<Int, String> {
 			val containerProperties = ContainerProperties(TEST_TOPIC5)
@@ -340,15 +345,15 @@ class KafkaDslKotlinTests {
 
 		@Bean
 		fun serverGateway() =
-				integrationFlow(Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) {
-					transform<String> { it.uppercase() }
-				}
+			integrationFlow(Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) {
+				transform<String> { it.uppercase() }
+			}
 
 		private fun containerProperties() =
-				ContainerProperties(TEST_TOPIC4)
-						.also {
-							it.setGroupId("inGateGroup")
-						}
+			ContainerProperties(TEST_TOPIC4)
+				.also {
+					it.setGroupId("inGateGroup")
+				}
 
 	}
 

spring-integration-kafka/src/test/resources/junit-platform.properties

@@ -0,0 +1,3 @@
+spring.kafka.global.embedded.enabled = true
+spring.embedded.kafka.brokers.property=spring.global.embedded.kafka.brokers
+spring.kafka.embedded.partitions=1