Upgrade to latest GAs

* Move all the test to JUnit 5; remove redundant JUnit 4 dependencies

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

@@ -21,8 +21,7 @@ import static org.mockito.Mockito.mock;
 
 import java.util.Collections;
 
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.channel.NullChannel;
@@ -40,16 +39,16 @@ import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
 import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.support.RetryTemplate;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Artem Bilan.
  * @author Gary Russell
  */
-@RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration
-public class KafkaMessageDrivenChannelAdapterParserTests {
+@SpringJUnitConfig
+@DirtiesContext
+class KafkaMessageDrivenChannelAdapterParserTests {
 
 	@Autowired
 	private NullChannel nullChannel;
@@ -73,7 +72,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 	private RecoveryCallback<?> recoveryCallback;
 
 	@Test
-	public void testKafkaMessageDrivenChannelAdapterParser() {
+	void testKafkaMessageDrivenChannelAdapterParser() {
 		assertThat(this.kafkaListener.isAutoStartup()).isFalse();
 		assertThat(this.kafkaListener.isRunning()).isFalse();
 		assertThat(this.kafkaListener.getPhase()).isEqualTo(100);
@@ -94,7 +93,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 	}
 
 	@Test
-	public void testKafkaBatchMessageDrivenChannelAdapterParser() {
+	void testKafkaBatchMessageDrivenChannelAdapterParser() {
 		assertThat(this.kafkaBatchListener.isAutoStartup()).isFalse();
 		assertThat(this.kafkaBatchListener.isRunning()).isFalse();
 		assertThat(this.kafkaBatchListener.getPhase()).isEqualTo(100);
@@ -110,7 +109,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests {
 
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testKafkaMessageDrivenChannelAdapterOptions() {
+	void testKafkaMessageDrivenChannelAdapterOptions() {
 		DefaultKafkaConsumerFactory<Integer, String> cf =
 				new DefaultKafkaConsumerFactory<>(Collections.emptyMap());
 		ContainerProperties containerProps = new ContainerProperties("foo");

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

@@ -17,7 +17,7 @@
 package org.springframework.integration.kafka.config.xml;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
@@ -29,8 +29,7 @@ import java.util.concurrent.TimeoutException;
 import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.common.serialization.IntegerSerializer;
 import org.apache.kafka.common.serialization.StringSerializer;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -45,8 +44,7 @@ import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.messaging.MessageHandlingException;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.ContextConfiguration;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Soby Chacko
@@ -57,18 +55,17 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  *
  * @since 0.5
  */
-@RunWith(SpringJUnit4ClassRunner.class)
-@ContextConfiguration
+@SpringJUnitConfig
 @DirtiesContext
-public class KafkaOutboundAdapterParserTests {
+class KafkaOutboundAdapterParserTests {
 
 	@Autowired
 	private ApplicationContext appContext;
 
 	@Test
-	public void testOutboundAdapterConfiguration() {
+	void testOutboundAdapterConfiguration() {
 		KafkaProducerMessageHandler<?, ?> messageHandler
-			= this.appContext.getBean("kafkaOutboundChannelAdapter.handler", KafkaProducerMessageHandler.class);
+				= this.appContext.getBean("kafkaOutboundChannelAdapter.handler", KafkaProducerMessageHandler.class);
 		assertThat(messageHandler).isNotNull();
 		assertThat(messageHandler.getOrder()).isEqualTo(3);
 		assertThat(TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue")).isEqualTo("foo");
@@ -98,7 +95,7 @@ public class KafkaOutboundAdapterParserTests {
 	}
 
 	@Test
-	public void testSyncMode() {
+	void testSyncMode() {
 		MockProducer<Integer, String> mockProducer =
 				new MockProducer<Integer, String>(false, new IntegerSerializer(), new StringSerializer()) {
 
@@ -128,28 +125,18 @@ public class KafkaOutboundAdapterParserTests {
 					return null;
 				});
 
-		try {
-			handler.handleMessage(new GenericMessage<>("foo"));
-			fail("MessageHandlingException expected");
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(MessageHandlingException.class);
-			assertThat(e.getCause()).isExactlyInstanceOf(KafkaProducerException.class);
-			assertThat(e.getCause().getCause()).isInstanceOf(RuntimeException.class);
-			assertThat(e.getMessage()).contains("Async Producer Mock exception");
-		}
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo")))
+				.withCauseInstanceOf(KafkaProducerException.class)
+				.withRootCauseInstanceOf(RuntimeException.class)
+				.withMessageContaining("Async Producer Mock exception");
 
 		handler.setSendTimeout(1);
 
-		try {
-			handler.handleMessage(new GenericMessage<>("foo"));
-			fail("MessageTimeoutException expected");
-		}
-		catch (Exception e) {
-			assertThat(e).isInstanceOf(MessageTimeoutException.class);
-			assertThat(e.getCause()).isExactlyInstanceOf(TimeoutException.class);
-			assertThat(e.getMessage()).contains("Timeout waiting for response from KafkaProducer");
-		}
+		assertThatExceptionOfType(MessageTimeoutException.class)
+				.isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo")))
+				.withCauseInstanceOf(TimeoutException.class)
+				.withMessageContaining("Timeout waiting for response from KafkaProducer");
 	}
 
 }

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

@@ -30,9 +30,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -67,7 +65,8 @@ 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.rule.EmbeddedKafkaRule;
+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;
@@ -78,7 +77,7 @@ import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.test.annotation.DirtiesContext;
-import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 
 /**
  * @author Artem Bilan
@@ -89,23 +88,21 @@ import org.springframework.test.context.junit4.SpringRunner;
  *
  * @since 3.0
  */
-@RunWith(SpringRunner.class)
+@SpringJUnitConfig
 @DirtiesContext
+@EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3,
+		KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5 })
 public class KafkaDslTests {
 
-	private static final String TEST_TOPIC1 = "test-topic1";
+	static final String TEST_TOPIC1 = "test-topic1";
 
-	private static final String TEST_TOPIC2 = "test-topic2";
+	static final String TEST_TOPIC2 = "test-topic2";
 
-	private static final String TEST_TOPIC3 = "test-topic3";
+	static final String TEST_TOPIC3 = "test-topic3";
 
-	private static final String TEST_TOPIC4 = "test-topic4";
+	static final String TEST_TOPIC4 = "test-topic4";
 
-	private static final String TEST_TOPIC5 = "test-topic5";
-
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafka =
-			new EmbeddedKafkaRule(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5);
+	static final String TEST_TOPIC5 = "test-topic5";
 
 	@Autowired
 	@Qualifier("sendToKafkaFlow.input")
@@ -151,7 +148,6 @@ public class KafkaDslTests {
 
 	@Test
 	public void testKafkaAdapters() throws Exception {
-
 		this.sendToKafkaFlowInput.send(new GenericMessage<>("foo", Collections.singletonMap("foo", "bar")));
 
 		assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper);
@@ -230,10 +226,14 @@ public class KafkaDslTests {
 
 		private Object fromSource;
 
+		@Autowired
+		private EmbeddedKafkaBroker embeddedKafka;
+
+
 		@Bean
 		public ConsumerFactory<Integer, String> consumerFactory() {
 			Map<String, Object> props = KafkaTestUtils
-					.consumerProps("test1", "false", embeddedKafka.getEmbeddedKafka());
+					.consumerProps("test1", "false", this.embeddedKafka);
 			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 			return new DefaultKafkaConsumerFactory<>(props);
 		}
@@ -295,7 +295,7 @@ public class KafkaDslTests {
 
 		@Bean
 		public ProducerFactory<Integer, String> producerFactory() {
-			Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
+			Map<String, Object> props = KafkaTestUtils.producerProps(this.embeddedKafka);
 			props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
 			return new DefaultKafkaProducerFactory<>(props);
 		}

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

@@ -29,9 +29,9 @@ import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -40,7 +40,6 @@ import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer;
 import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
 import org.springframework.integration.support.MessageBuilder;
-import org.springframework.integration.test.rule.Log4j2LevelAdjuster;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
@@ -52,7 +51,6 @@ import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.ContainerTestUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
@@ -76,7 +74,7 @@ import org.springframework.retry.support.RetryTemplate;
  * @since 3.0.2
  *
  */
-public class InboundGatewayTests {
+class InboundGatewayTests {
 
 	private static String topic1 = "testTopic1";
 
@@ -92,32 +90,37 @@ public class InboundGatewayTests {
 
 	private static String topic7 = "testTopic7";
 
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafka =
-			new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6, topic7);
+	private static EmbeddedKafkaBroker embeddedKafka;
 
-	@Rule
-	public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace().categories("org.apache.kafka.clients",
-			"org.springframework.kafka", "org.springframework.integration");
+	@BeforeAll
+	static void setup() {
+		embeddedKafka = new EmbeddedKafkaBroker(1, true,
+				topic1, topic2, topic3, topic4, topic5, topic6, topic7);
+		embeddedKafka.afterPropertiesSet();
+	}
+
+	@AfterAll
+	static void tearDown() {
+		embeddedKafka.destroy();
+	}
 
 	@Test
-	public void testInbound() throws Exception {
-		EmbeddedKafkaBroker embedded = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
+	void testInbound() throws Exception {
 		Map<String, Object> consumerProps =
-				KafkaTestUtils.consumerProps("replyHandler1", "false", embedded);
+				KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
 		Consumer<Integer, String> consumer = cf2.createConsumer();
-		embedded.consumeFromAnEmbeddedTopic(consumer, topic2);
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic2);
 
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embedded);
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic1);
 		containerProps.setIdleEventInterval(100L);
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embedded);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic1);
@@ -177,21 +180,20 @@ public class InboundGatewayTests {
 	}
 
 	@Test
-	public void testInboundErrorRecover() {
-		EmbeddedKafkaBroker broker = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
-		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", broker);
+	void testInboundErrorRecover() {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", embeddedKafka);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
 		Consumer<Integer, String> consumer = cf2.createConsumer();
-		broker.consumeFromAnEmbeddedTopic(consumer, topic4);
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic4);
 
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "false", broker);
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic3);
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(broker);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic3);
@@ -257,21 +259,20 @@ public class InboundGatewayTests {
 	}
 
 	@Test
-	public void testInboundRetryErrorRecover() {
-		EmbeddedKafkaBroker embedded = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
-		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embedded);
+	void testInboundRetryErrorRecover() {
+		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embeddedKafka);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);
 		Consumer<Integer, String> consumer = cf2.createConsumer();
-		embedded.consumeFromAnEmbeddedTopic(consumer, topic6);
+		embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic6);
 
-		Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "false", embedded);
+		Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic5);
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
-		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embedded);
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(topic5);
@@ -342,8 +343,7 @@ public class InboundGatewayTests {
 	}
 
 	@Test
-	public void testInboundRetryErrorRecoverWithoutRecocveryCallback() throws Exception {
-		EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka();
+	void testInboundRetryErrorRecoverWithoutRecocveryCallback() throws Exception {
 		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("replyHandler4", "false", embeddedKafka);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		ConsumerFactory<Integer, String> cf2 = new DefaultKafkaConsumerFactory<>(consumerProps);

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

@@ -46,8 +46,9 @@ 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.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.StaticMessageHeaderAccessor;
@@ -77,7 +78,6 @@ 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.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.ContainerTestUtils;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
@@ -102,7 +102,7 @@ import org.springframework.retry.support.RetryTemplate;
  * @since 2.0
  *
  */
-public class MessageDrivenAdapterTests {
+class MessageDrivenAdapterTests {
 
 	private static String topic1 = "testTopic1";
 
@@ -116,14 +116,22 @@ public class MessageDrivenAdapterTests {
 
 	private static String topic6 = "testTopic6";
 
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafkaRule =
-			new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6);
+	private static EmbeddedKafkaBroker embeddedKafka;
 
-	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
+	@BeforeAll
+	static void setup() {
+		embeddedKafka = new EmbeddedKafkaBroker(1, true,
+				topic1, topic2, topic3, topic4, topic5, topic6);
+		embeddedKafka.afterPropertiesSet();
+	}
+
+	@AfterAll
+	static void tearDown() {
+		embeddedKafka.destroy();
+	}
 
 	@Test
-	public void testInboundRecord() {
+	void testInboundRecord() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -208,7 +216,7 @@ public class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	public void testInboundRecordRetryRecover() {
+	void testInboundRecordRetryRecover() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -266,7 +274,7 @@ public class MessageDrivenAdapterTests {
 	 * to the consumer.
 	 */
 	@Test
-	public void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception {
+	void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -316,7 +324,7 @@ public class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	public void testInboundRecordNoRetryRecover() {
+	void testInboundRecordNoRetryRecover() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -368,7 +376,7 @@ public class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	public void testInboundBatch() throws Exception {
+	void testInboundBatch() throws Exception {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -391,6 +399,7 @@ public class MessageDrivenAdapterTests {
 			@Override
 			public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment,
 					Consumer<?, ?> consumer, Type type) {
+
 				Message<?> message = super.toMessage(records, acknowledgment, consumer, type);
 				return MessageBuilder.fromMessage(message).setHeader("testHeader", "testValue").build();
 			}
@@ -431,6 +440,7 @@ public class MessageDrivenAdapterTests {
 			@Override
 			public Message<?> toMessage(List<ConsumerRecord<?, ?>> records, Acknowledgment acknowledgment,
 					Consumer<?, ?> consumer, Type payloadType) {
+
 				throw new RuntimeException("testError");
 			}
 
@@ -453,7 +463,7 @@ public class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	public void testInboundJson() {
+	void testInboundJson() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -495,7 +505,7 @@ public class MessageDrivenAdapterTests {
 	}
 
 	@Test
-	public void testInboundJsonWithPayload() {
+	void testInboundJsonWithPayload() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka);
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -503,7 +513,8 @@ public class MessageDrivenAdapterTests {
 		KafkaMessageListenerContainer<Integer, Foo> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 
-		KafkaMessageDrivenChannelAdapter<Integer, Foo> adapter = Kafka.messageDrivenChannelAdapter(container, ListenerMode.record)
+		KafkaMessageDrivenChannelAdapter<Integer, Foo> adapter = Kafka
+				.messageDrivenChannelAdapter(container, ListenerMode.record)
 				.recordMessageConverter(new StringJsonMessageConverter())
 				.payloadType(Foo.class)
 				.get();
@@ -542,7 +553,7 @@ public class MessageDrivenAdapterTests {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	public void testPauseResume() throws Exception {
+	void testPauseResume() throws Exception {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
 		Consumer<Integer, String> consumer = mock(Consumer.class);
 		given(cf.createConsumer(isNull(), eq("clientId"), isNull(), any())).willReturn(consumer);
@@ -597,14 +608,14 @@ public class MessageDrivenAdapterTests {
 		adapter.stop();
 	}
 
-	public static class Foo {
+	static class Foo {
 
 		private String bar;
 
-		public Foo() {
+		Foo() {
 		}
 
-		public Foo(String bar) {
+		Foo(String bar) {
 			this.bar = bar;
 		}
 
@@ -637,14 +648,11 @@ public class MessageDrivenAdapterTests {
 			}
 			Foo other = (Foo) obj;
 			if (this.bar == null) {
-				if (other.bar != null) {
-					return false;
-				}
+				return other.bar == null;
 			}
-			else if (!this.bar.equals(other.bar)) {
-				return false;
+			else {
+				return this.bar.equals(other.bar);
 			}
-			return true;
 		}
 
 	}

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

@@ -26,15 +26,15 @@ import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.common.TopicPartition;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 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.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
 
@@ -46,17 +46,25 @@ import org.springframework.messaging.Message;
  * @since 3.0.1
  *
  */
-public class MessageSourceIntegrationTests {
+class MessageSourceIntegrationTests {
 
-	public static final String TOPIC1 = "MessageSourceIntegrationTests1";
+	private static final String TOPIC1 = "MessageSourceIntegrationTests1";
 
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, 1, TOPIC1);
+	private static EmbeddedKafkaBroker embeddedKafka;
 
-	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
+	@BeforeAll
+	static void setup() {
+		embeddedKafka = new EmbeddedKafkaBroker(1, true, 1, TOPIC1);
+		embeddedKafka.afterPropertiesSet();
+	}
+
+	@AfterAll
+	static void tearDown() {
+		embeddedKafka.destroy();
+	}
 
 	@Test
-	public void testSource() throws Exception {
+	void testSource() throws Exception {
 		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testSource", "false", embeddedKafka);
 		consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
@@ -64,7 +72,7 @@ public class MessageSourceIntegrationTests {
 		DefaultKafkaConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps);
 		ConsumerProperties consumerProperties = new ConsumerProperties(TOPIC1);
 		consumerProperties.getKafkaConsumerProperties()
-			.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "2");
+				.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "2");
 		final CountDownLatch assigned = new CountDownLatch(1);
 		consumerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
 

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

@@ -17,8 +17,8 @@
 package org.springframework.integration.kafka.inbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyCollection;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -64,7 +64,7 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.TimestampType;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 
@@ -87,14 +87,16 @@ import org.springframework.messaging.Message;
 /**
  * @author Gary Russell
  * @author Anshul Mehra
+ * @author Artem Bilan
+ *
  * @since 3.0.1
  *
  */
-public class MessageSourceTests {
+class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testIllegalArgs() {
+	void testIllegalArgs() {
 		ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
 		assertThatThrownBy(() -> new KafkaMessageSource(consumerFactory, new ConsumerProperties((Pattern) null)))
 				.isInstanceOf(IllegalArgumentException.class)
@@ -103,7 +105,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	public void testConsumerAwareRebalanceListener() {
+	void testConsumerAwareRebalanceListener() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		List<TopicPartition> assigned = Collections.singletonList(topicPartition);
@@ -123,6 +125,7 @@ public class MessageSourceTests {
 		AtomicBoolean partitionsRevokedCalled = new AtomicBoolean();
 		AtomicReference<Consumer> partitionsRevokedConsumer = new AtomicReference<>();
 		consumerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
+
 			@Override
 			public void onPartitionsRevokedAfterCommit(Consumer<?, ?> cons, Collection<TopicPartition> partitions) {
 				partitionsRevokedCalled.getAndSet(true);
@@ -152,7 +155,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	public void testRebalanceListener() {
+	void testRebalanceListener() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		List<TopicPartition> assigned = Collections.singletonList(topicPartition);
@@ -195,17 +198,17 @@ public class MessageSourceTests {
 	}
 
 	@Test
-	public void testAckSyncCommits() {
+	void testAckSyncCommits() {
 		testAckCommon(true, false);
 	}
 
 	@Test
-	public void testAckSyncCommitsTimeout() {
+	void testAckSyncCommitsTimeout() {
 		testAckCommon(true, false);
 	}
 
 	@Test
-	public void testAckAsyncCommits() {
+	void testAckAsyncCommits() {
 		testAckCommon(false, false);
 	}
 
@@ -269,7 +272,7 @@ public class MessageSourceTests {
 		assertThat(received).isNotNull();
 		assertThat(received.getHeaders().get(KafkaHeaders.RAW_DATA)).isInstanceOf(ConsumerRecord.class);
 		assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA))
-			.isSameAs(received.getHeaders().get(KafkaHeaders.RAW_DATA));
+				.isSameAs(received.getHeaders().get(KafkaHeaders.RAW_DATA));
 		StaticMessageHeaderAccessor.getAcknowledgmentCallback(received)
 				.acknowledge(AcknowledgmentCallback.Status.ACCEPT);
 		received = source.receive();
@@ -337,7 +340,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testAckOutOfOrder() {
+	void testAckOutOfOrder() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -427,7 +430,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testNack() {
+	void testNack() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -442,7 +445,7 @@ public class MessageSourceTests {
 		}).given(consumer).pause(anyCollection());
 		willAnswer(i -> paused.get()).given(consumer).paused();
 		Map<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
-		records1.put(topicPartition, Arrays.asList(
+		records1.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
 		ConsumerRecords cr1 = new ConsumerRecords(records1);
 		Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
@@ -496,7 +499,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testNackWithLaterInflight() {
+	void testNackWithLaterInflight() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -511,7 +514,7 @@ public class MessageSourceTests {
 		}).given(consumer).pause(anyCollection());
 		willAnswer(i -> paused.get()).given(consumer).paused();
 		Map<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
-		records1.put(topicPartition, Arrays.asList(
+		records1.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
 		ConsumerRecords cr1 = new ConsumerRecords(records1);
 		Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
@@ -582,7 +585,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testMaxPollRecords() {
+	void testMaxPollRecords() {
 		KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()),
 				new ConsumerProperties("topic"));
 		assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class)
@@ -591,20 +594,17 @@ public class MessageSourceTests {
 				Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)), new ConsumerProperties("topic"));
 		assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class)
 				.get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1);
-		try {
-			new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) {
 
-			}), new ConsumerProperties("topic"));
-			fail("Expected exception");
-		}
-		catch (IllegalArgumentException e) {
-			assertThat(e.getMessage()).contains(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
-		}
+		assertThatIllegalArgumentException()
+				.isThrownBy(() ->
+						new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) { }),
+								new ConsumerProperties("topic")))
+				.withMessageContaining(ConsumerConfig.MAX_POLL_RECORDS_CONFIG);
 	}
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testPollTimeouts() {
+	void testPollTimeouts() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		List<TopicPartition> assigned = Collections.singletonList(topicPartition);
@@ -615,12 +615,12 @@ public class MessageSourceTests {
 		}).given(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class));
 
 		Map<TopicPartition, List<ConsumerRecord>> records1 = new LinkedHashMap<>();
-		records1.put(topicPartition, Arrays.asList(
+		records1.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
 		ConsumerRecords cr1 = new ConsumerRecords(records1);
 		given(consumer.poll(Duration.of(20 * 5000, ChronoUnit.MILLIS))).willReturn(cr1, ConsumerRecords.EMPTY);
 		Map<TopicPartition, List<ConsumerRecord>> records2 = new LinkedHashMap<>();
-		records2.put(topicPartition, Arrays.asList(
+		records2.put(topicPartition, Collections.singletonList(
 				new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo")));
 		ConsumerRecords cr2 = new ConsumerRecords(records2);
 		given(consumer.poll(Duration.of(5000, ChronoUnit.MILLIS))).willReturn(cr2, ConsumerRecords.EMPTY);
@@ -665,7 +665,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testAllowMulti() {
+	void testAllowMulti() {
 		Consumer consumer = mock(Consumer.class);
 		TopicPartition topicPartition = new TopicPartition("foo", 0);
 		List<TopicPartition> assigned = Collections.singletonList(topicPartition);
@@ -730,7 +730,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings("unchecked")
 	@Test
-	public void testTopicPatternBasedMessageSource() {
+	void testTopicPatternBasedMessageSource() {
 		MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
 		TopicPartition topicPartition1 = new TopicPartition("abc_foo", 0);
 		TopicPartition topicPartition2 = new TopicPartition("abc_foo", 1);
@@ -747,7 +747,9 @@ public class MessageSourceTests {
 		willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory)
 				.getConfigurationProperties();
 		given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
-		KafkaMessageSource<String, String> source = new KafkaMessageSource<>(consumerFactory, new ConsumerProperties(Pattern.compile("[a-zA-Z0-9_]*?foo")));
+		KafkaMessageSource<String, String> source = new KafkaMessageSource<>(consumerFactory,
+				new ConsumerProperties(Pattern
+						.compile("[a-zA-Z0-9_]*?foo")));
 		source.setRawMessageHeader(true);
 		source.start();
 		// force consumer creation
@@ -789,7 +791,7 @@ public class MessageSourceTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testStaticPartitionAssignment() {
+	void testStaticPartitionAssignment() {
 		MockConsumer<String, String> consumer = spy(new MockConsumer<>(OffsetResetStrategy.EARLIEST));
 
 		TopicPartition beginning = new TopicPartition("foo", 0);
@@ -818,9 +820,11 @@ public class MessageSourceTests {
 				.getConfigurationProperties();
 		given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer);
 
-		TopicPartitionOffset beginningTpo = new TopicPartitionOffset(beginning, null, TopicPartitionOffset.SeekPosition.BEGINNING);
+		TopicPartitionOffset beginningTpo = new TopicPartitionOffset(beginning, null,
+				TopicPartitionOffset.SeekPosition.BEGINNING);
 		TopicPartitionOffset endTpo = new TopicPartitionOffset(end, null, TopicPartitionOffset.SeekPosition.END);
-		TopicPartitionOffset timestampTpo = new TopicPartitionOffset(timestamp, null, TopicPartitionOffset.SeekPosition.TIMESTAMP);
+		TopicPartitionOffset timestampTpo = new TopicPartitionOffset(timestamp, null,
+				TopicPartitionOffset.SeekPosition.TIMESTAMP);
 		TopicPartitionOffset negativeOffsetTpo = new TopicPartitionOffset(negativeOffset, -1L, null);
 		TopicPartitionOffset negativeRelativeToCurrentTpo = new TopicPartitionOffset(negativeRelativeToCurrent.topic(),
 				negativeRelativeToCurrent.partition(), -1L, true);
@@ -865,11 +869,11 @@ public class MessageSourceTests {
 		Message<Object> message;
 		Set<String> expected = Stream.of(
 				p0r0, p0r1, p0r2, p0r3, // Seek to beginning
-				p1r3, 					// Seek to end
-				p2r1, p2r2, p2r3, 		// Null offset and SeekPosition.TIMESTAMP results in no change in position
-				p3r3, 					// Negative offset ends up in seek to end
-				p4r2, p4r3, 			// Negative offset with relative to current(3)
-				p5r2, p5r3				// Positive offset with relative to current(1)
+				p1r3,                    // Seek to end
+				p2r1, p2r2, p2r3,        // Null offset and SeekPosition.TIMESTAMP results in no change in position
+				p3r3,                    // Negative offset ends up in seek to end
+				p4r2, p4r3,            // Negative offset with relative to current(3)
+				p5r2, p5r3                // Positive offset with relative to current(1)
 		).map(ConsumerRecord::value).collect(Collectors.toSet());
 		Set<Object> received = new HashSet<>();
 		while ((message = source.receive()) != null) {
@@ -902,4 +906,5 @@ public class MessageSourceTests {
 		inOrder.verify(consumer).close(anyLong(), any(TimeUnit.class));
 		inOrder.verifyNoMoreInteractions();
 	}
+
 }

spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java

@@ -17,7 +17,7 @@
 package org.springframework.integration.kafka.outbound;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.fail;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.BDDMockito.given;
@@ -56,10 +56,9 @@ import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.TopicPartition;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 
@@ -87,7 +86,6 @@ import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.support.TransactionSupport;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.kafka.transaction.KafkaTransactionManager;
 import org.springframework.messaging.Message;
@@ -108,7 +106,7 @@ import org.springframework.util.concurrent.SettableListenableFuture;
  *
  * @since 2.0
  */
-public class KafkaProducerMessageHandlerTests {
+class KafkaProducerMessageHandlerTests {
 
 	private static String topic1 = "testTopic1out";
 
@@ -122,16 +120,15 @@ public class KafkaProducerMessageHandlerTests {
 
 	private static String topic6 = "testTopic6in";
 
-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafkaRule =
-			new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6);
-
-	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
+	private static EmbeddedKafkaBroker embeddedKafka;
 
 	private static Consumer<Integer, String> consumer;
 
-	@BeforeClass
-	public static void setUp() {
+	@BeforeAll
+	static void setup() {
+		embeddedKafka = new EmbeddedKafkaBroker(1, true,
+				topic1, topic2, topic3, topic4, topic5, topic6);
+		embeddedKafka.afterPropertiesSet();
 		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testOut", "true", embeddedKafka);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
@@ -139,13 +136,14 @@ public class KafkaProducerMessageHandlerTests {
 		embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
 	}
 
-	@AfterClass
-	public static void tearDown() {
+	@AfterAll
+	static void tearDown() {
 		consumer.close();
+		embeddedKafka.destroy();
 	}
 
 	@Test
-	public void testOutbound() throws Exception {
+	void testOutbound() {
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
@@ -171,7 +169,7 @@ public class KafkaProducerMessageHandlerTests {
 				.build();
 		handler.handleMessage(message);
 		record = KafkaTestUtils.getSingleRecord(consumer, topic1);
-		assertThat(record).has(key((Integer) null));
+		assertThat(record).has(key(null));
 		assertThat(record).has(partition(0));
 		assertThat(record).has(value("bar"));
 
@@ -180,7 +178,7 @@ public class KafkaProducerMessageHandlerTests {
 				.build();
 		handler.handleMessage(message);
 		record = KafkaTestUtils.getSingleRecord(consumer, topic1);
-		assertThat(record).has(key((Integer) null));
+		assertThat(record).has(key(null));
 		assertThat(record).has(value("baz"));
 
 		handler.setPartitionIdExpression(new SpelExpressionParser().parseExpression("headers['kafka_partitionId']"));
@@ -201,7 +199,7 @@ public class KafkaProducerMessageHandlerTests {
 	}
 
 	@Test
-	public void testOutboundWithTimestamp() {
+	void testOutboundWithTimestamp() {
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
@@ -232,7 +230,7 @@ public class KafkaProducerMessageHandlerTests {
 	}
 
 	@Test
-	public void testOutboundWithTimestampExpression() {
+	void testOutboundWithTimestampExpression() {
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
@@ -271,7 +269,7 @@ public class KafkaProducerMessageHandlerTests {
 	}
 
 	@Test
-	public void testOutboundWithAsyncResults() {
+	void testOutboundWithAsyncResults() {
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
@@ -331,7 +329,7 @@ public class KafkaProducerMessageHandlerTests {
 	}
 
 	@Test
-	public void testOutboundWithCustomHeaderMapper() throws Exception {
+	void testOutboundWithCustomHeaderMapper() {
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
@@ -353,7 +351,7 @@ public class KafkaProducerMessageHandlerTests {
 	}
 
 	@Test
-	public void testOutboundGateway() throws Exception {
+	void testOutboundGateway() throws Exception {
 		ConsumerFactory<Integer, String> consumerFactory = new DefaultKafkaConsumerFactory<>(
 				KafkaTestUtils.consumerProps(topic5, "false", embeddedKafka));
 		ContainerProperties containerProperties = new ContainerProperties(topic6);
@@ -376,7 +374,8 @@ public class KafkaProducerMessageHandlerTests {
 
 		DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
 				KafkaTestUtils.producerProps(embeddedKafka));
-		ReplyingKafkaTemplate<Integer, String, String> template = new ReplyingKafkaTemplate<>(producerFactory, container);
+		ReplyingKafkaTemplate<Integer, String, String> template =
+				new ReplyingKafkaTemplate<>(producerFactory, container);
 		template.start();
 		assertThat(assigned.await(30, TimeUnit.SECONDS)).isTrue();
 		KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
@@ -407,37 +406,30 @@ public class KafkaProducerMessageHandlerTests {
 		assertThat(reply.getHeaders().get(KafkaHeaders.TOPIC)).isNull();
 		assertThat(reply.getHeaders().get(KafkaHeaders.CORRELATION_ID)).isNull();
 
-		message = MessageBuilder.withPayload("foo")
+		final Message<?> messageToHandle1 = MessageBuilder.withPayload("foo")
 				.setHeader(KafkaHeaders.TOPIC, topic5)
 				.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
 				.setHeader(KafkaHeaders.PARTITION_ID, 1)
 				.setHeader(KafkaHeaders.REPLY_TOPIC, "bad")
 				.build();
-		try {
-			handler.handleMessage(message);
-			fail("Expected exception");
-		}
-		catch (MessageHandlingException e) {
-			assertThat(e.getCause().getMessage())
-					.isEqualTo("The reply topic header [bad] does not match any reply container topic: "
-							+ "[" + topic6 + "]");
-		}
-		message = MessageBuilder.withPayload("foo")
+
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> handler.handleMessage(messageToHandle1))
+				.withMessageContaining("The reply topic header [bad] does not match any reply container topic: "
+						+ "[" + topic6 + "]");
+
+		final Message<?> messageToHandle2 = MessageBuilder.withPayload("foo")
 				.setHeader(KafkaHeaders.TOPIC, topic5)
 				.setHeader(KafkaHeaders.MESSAGE_KEY, 2)
 				.setHeader(KafkaHeaders.PARTITION_ID, 1)
 				.setHeader(KafkaHeaders.REPLY_PARTITION, 999)
 				.build();
-		try {
-			handler.handleMessage(message);
-			fail("Expected exception");
-		}
-		catch (MessageHandlingException e) {
-			assertThat(e.getCause().getMessage())
-					.isEqualTo(
-							"The reply partition header [999] does not match any reply container partition for topic ["
-									+ topic6 + "]: [0, 1]");
-		}
+
+		assertThatExceptionOfType(MessageHandlingException.class)
+				.isThrownBy(() -> handler.handleMessage(messageToHandle2))
+				.withMessageContaining("The reply partition header [999] " +
+						"does not match any reply container partition for topic ["
+						+ topic6 + "]: [0, 1]");
 
 		template.stop();
 		// discard from the test consumer
@@ -448,7 +440,7 @@ public class KafkaProducerMessageHandlerTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testTransaction() {
+	void testTransaction() {
 		ProducerFactory pf = mock(ProducerFactory.class);
 		given(pf.transactionCapable()).willReturn(true);
 		Producer producer = mock(Producer.class);
@@ -472,7 +464,7 @@ public class KafkaProducerMessageHandlerTests {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	public void testConsumeAndProduceTransaction() throws Exception {
+	void testConsumeAndProduceTransaction() throws Exception {
 		Consumer mockConsumer = mock(Consumer.class);
 		final TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {
@@ -507,8 +499,7 @@ public class KafkaProducerMessageHandlerTests {
 			transactionalIds.add(TransactionSupport.getTransactionIdSuffix());
 			return producer;
 		}).given(pf).createProducer(isNull());
-		KafkaTransactionManager tm = new KafkaTransactionManager(pf);
-		PlatformTransactionManager ptm = tm;
+		PlatformTransactionManager ptm = new KafkaTransactionManager(pf);
 		ContainerProperties props = new ContainerProperties("foo");
 		props.setGroupId("group");
 		props.setTransactionManager(ptm);
@@ -549,7 +540,7 @@ public class KafkaProducerMessageHandlerTests {
 
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testTransactionTxIdOverride() {
+	void testTransactionTxIdOverride() {
 		Producer producer = mock(Producer.class);
 		AtomicReference<String> txId = new AtomicReference<>();
 		DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(Collections.emptyMap()) {
@@ -584,7 +575,7 @@ public class KafkaProducerMessageHandlerTests {
 
 	@SuppressWarnings({ "unchecked", "rawtypes" })
 	@Test
-	public void testConsumeAndProduceTransactionTxIdOverride() throws Exception {
+	void testConsumeAndProduceTransactionTxIdOverride() throws Exception {
 		Consumer mockConsumer = mock(Consumer.class);
 		final TopicPartition topicPartition = new TopicPartition("foo", 0);
 		willAnswer(i -> {

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

@@ -27,9 +27,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener
 import org.apache.kafka.clients.producer.ProducerConfig
 import org.apache.kafka.common.TopicPartition
-import org.junit.ClassRule
 import org.junit.Test
-import org.junit.runner.RunWith
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.beans.factory.annotation.Qualifier
 import org.springframework.context.annotation.Bean
@@ -62,7 +60,8 @@ 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.rule.EmbeddedKafkaRule
+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,7 +71,7 @@ import org.springframework.messaging.support.ErrorMessage
 import org.springframework.messaging.support.GenericMessage
 import org.springframework.retry.support.RetryTemplate
 import org.springframework.test.annotation.DirtiesContext
-import org.springframework.test.context.junit4.SpringRunner
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig
 import java.time.Duration
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
@@ -85,296 +84,298 @@ import java.util.stream.Stream
  * @since 3.0.3
  */
 
-@RunWith(SpringRunner::class)
+@SpringJUnitConfig
 @DirtiesContext
+@EmbeddedKafka(topics = [KafkaDslKotlinTests.TEST_TOPIC1, KafkaDslKotlinTests.TEST_TOPIC2,
+    KafkaDslKotlinTests.TEST_TOPIC3, KafkaDslKotlinTests.TEST_TOPIC4, KafkaDslKotlinTests.TEST_TOPIC5])
 class KafkaDslKotlinTests {
 
-    companion object {
+	companion object {
 
-        const val TEST_TOPIC1 = "test-topic1"
+		const val TEST_TOPIC1 = "test-topic1"
 
-        const val TEST_TOPIC2 = "test-topic2"
+		const val TEST_TOPIC2 = "test-topic2"
 
-        const val TEST_TOPIC3 = "test-topic3"
+		const val TEST_TOPIC3 = "test-topic3"
 
-        const val TEST_TOPIC4 = "test-topic4"
+		const val TEST_TOPIC4 = "test-topic4"
 
-        const val TEST_TOPIC5 = "test-topic5"
+		const val TEST_TOPIC5 = "test-topic5"
+
+	}
 
 
-        @ClassRule
-        @JvmField
-        var embeddedKafka = EmbeddedKafkaRule(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5)
-    }
+	@Autowired
+	@Qualifier("sendToKafkaFlow.input")
+	private lateinit var sendToKafkaFlowInput: MessageChannel
+
+	@Autowired
+	private lateinit var listeningFromKafkaResults1: PollableChannel
+
+	@Autowired
+	private lateinit var listeningFromKafkaResults2: PollableChannel
+
+	@Autowired
+	@Qualifier("kafkaProducer1.handler")
+	private lateinit var kafkaProducer1: KafkaProducerMessageHandler<*, *>
+
+	@Autowired
+	@Qualifier("kafkaProducer2.handler")
+	private lateinit var kafkaProducer2: KafkaProducerMessageHandler<*, *>
+
+	@Autowired
+	private lateinit var errorChannel: PollableChannel
+
+	@Autowired(required = false)
+	@Qualifier("topic1ListenerContainer")
+	private lateinit var messageListenerContainer: MessageListenerContainer
+
+	@Autowired(required = false)
+	@Qualifier("kafkaTemplate:test-topic1")
+	private lateinit var kafkaTemplateTopic1: KafkaTemplate<Any, Any>
+
+	@Autowired(required = false)
+	@Qualifier("kafkaTemplate:test-topic2")
+	private lateinit var kafkaTemplateTopic2: KafkaTemplate<*, *>
+
+	@Autowired
+	private lateinit var mapper: DefaultKafkaHeaderMapper
+
+	@Autowired
+	private lateinit var config: ContextConfiguration
+
+	@Autowired
+	private lateinit var gate: Gate
+
+	@Test
+	fun testKafkaAdapters() {
+		this.sendToKafkaFlowInput.send(GenericMessage("foo", hashMapOf<String, Any>("foo" to "bar")))
+
+		assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper)
+
+		for (i in 0..99) {
+			val receive = this.listeningFromKafkaResults1.receive(20000)
+			assertThat(receive).isNotNull()
+			assertThat(receive!!.payload).isEqualTo("FOO")
+			val headers = receive.headers
+			assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue()
+			val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java)
+			acknowledgment?.acknowledge()
+			assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC1)
+			assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1)
+			assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0)
+			assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong())
+			assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME")
+			assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048633L)
+			assertThat(headers["foo"]).isEqualTo("bar")
+		}
+
+		for (i in 0..99) {
+			val receive = this.listeningFromKafkaResults2.receive(20000)
+			assertThat(receive).isNotNull()
+			assertThat(receive!!.payload).isEqualTo("FOO")
+			val headers = receive.headers
+			assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue()
+			val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java)
+			acknowledgment?.acknowledge()
+			assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC2)
+			assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1)
+			assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0)
+			assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong())
+			assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME")
+			assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048644L)
+		}
+
+		val message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build()
+
+		this.sendToKafkaFlowInput.send(message)
+
+		assertThat(this.listeningFromKafkaResults1.receive(10)).isNull()
+
+		val error = this.errorChannel.receive(10000)
+
+		assertThat(error).isNotNull().isInstanceOf(ErrorMessage::class.java)
+
+		val payload = error?.payload
+
+		assertThat(payload).isNotNull().isInstanceOf(MessageRejectedException::class.java)
+
+		assertThat(this.messageListenerContainer).isNotNull()
+		assertThat(this.kafkaTemplateTopic1).isNotNull()
+		assertThat(this.kafkaTemplateTopic2).isNotNull()
+
+		this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo")
+		assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue()
+		assertThat(this.config.fromSource).isEqualTo("foo")
+	}
+
+	@Test
+	fun testGateways() {
+		assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS))
+		assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO")
+	}
+
+	@Configuration
+	@EnableIntegration
+	@EnableKafka
+	class ContextConfiguration {
+
+		val sourceFlowLatch = CountDownLatch(1)
+
+		val replyContainerLatch = CountDownLatch(1)
+
+		var fromSource: Any? = null
+
+		@Autowired
+		private lateinit var embeddedKafka: EmbeddedKafkaBroker
+
+		@Bean
+		fun consumerFactory(): ConsumerFactory<Int, String> {
+			val props = KafkaTestUtils.consumerProps("test1", "false", this.embeddedKafka)
+			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+			return DefaultKafkaConsumerFactory(props)
+		}
+
+		@Bean
+		fun errorChannel() = QueueChannel()
+
+		@Bean
+		fun topic1ListenerFromKafkaFlow() =
+				IntegrationFlows.from(
+						Kafka.messageDrivenChannelAdapter(consumerFactory(),
+								KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
+								.configureListenerContainer {
+									it.ackMode(ContainerProperties.AckMode.MANUAL)
+											.id("topic1ListenerContainer")
+								}
+								.recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(),
+										RawRecordHeaderErrorMessageStrategy()))
+								.retryTemplate(RetryTemplate())
+								.filterInRetry(true))
+						.filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 },
+								{ f -> f.throwExceptionOnRejection(true) })
+						.transform<String, String> { it.toUpperCase() }
+						.channel { c -> c.queue("listeningFromKafkaResults1") }
+						.get()
+
+		@Bean
+		fun topic2ListenerFromKafkaFlow() =
+				IntegrationFlows.from(
+						Kafka.messageDrivenChannelAdapter(consumerFactory(),
+								KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2)
+								.configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) }
+								.recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(),
+										RawRecordHeaderErrorMessageStrategy()))
+								.retryTemplate(RetryTemplate())
+								.filterInRetry(true))
+						.filter(Message::class.java,
+								{ m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 },
+								{ it.throwExceptionOnRejection(true) })
+						.transform<String, String> { it.toUpperCase() }
+						.channel { c -> c.queue("listeningFromKafkaResults2") }
+						.get()
+
+		@Bean
+		fun producerFactory(): DefaultKafkaProducerFactory<Int, String> {
+			val props = KafkaTestUtils.producerProps(this.embeddedKafka)
+			props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000")
+			return DefaultKafkaProducerFactory(props)
+		}
+
+		@Bean
+		fun sendToKafkaFlow() =
+				IntegrationFlow { f ->
+					f.split<String>({ p -> Stream.generate { p }.limit(101) }, null)
+							.publishSubscribeChannel { c ->
+								c.subscribe { sf ->
+									sf.handle(
+											kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
+													.timestampExpression("T(Long).valueOf('1487694048633')")
+									) { it.id("kafkaProducer1") }
+								}
+										.subscribe { sf ->
+											sf.handle(
+													kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
+															.timestamp<Any> { 1487694048644L }
+											) { it.id("kafkaProducer2") }
+										}
+							}
+				}
+
+		@Bean
+		fun mapper() = DefaultKafkaHeaderMapper()
+
+		private fun kafkaMessageHandler(producerFactory: ProducerFactory<Int, String>, topic: String) =
+				Kafka.outboundChannelAdapter(producerFactory)
+						.messageKey<Any> { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
+						.headerMapper(mapper())
+						.sync(true)
+						.partitionId<Any> { _ -> 0 }
+						.topicExpression("headers[kafka_topic] ?: '$topic'")
+						.configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") }
 
 
-    @Autowired
-    @Qualifier("sendToKafkaFlow.input")
-    private lateinit var sendToKafkaFlowInput: MessageChannel
+		@Bean
+		fun sourceFlow() =
+				IntegrationFlows
+						.from(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)))
+						{ e -> e.poller(Pollers.fixedDelay(100)) }
+						.handle { p ->
+							this.fromSource = p.getPayload()
+							this.sourceFlowLatch.countDown()
+						}
+						.get()
 
-    @Autowired
-    private lateinit var listeningFromKafkaResults1: PollableChannel
+		@Bean
+		fun replyingKafkaTemplate() =
+				ReplyingKafkaTemplate(producerFactory(), replyContainer())
+						.also {
+							it.setDefaultReplyTimeout(Duration.ofSeconds(30))
+						}
 
-    @Autowired
-    private lateinit var listeningFromKafkaResults2: PollableChannel
+		@Bean
+		fun outboundGateFlow() =
+				IntegrationFlows.from(Gate::class.java)
+						.handle(Kafka.outboundGateway(replyingKafkaTemplate())
+								.sync(true))
+						.get()
 
-    @Autowired
-    @Qualifier("kafkaProducer1.handler")
-    private lateinit var kafkaProducer1: KafkaProducerMessageHandler<*, *>
+		private fun replyContainer(): GenericMessageListenerContainer<Int, String> {
+			val containerProperties = ContainerProperties(TEST_TOPIC5)
+			containerProperties.groupId = "outGate"
+			containerProperties.consumerRebalanceListener = object : ConsumerRebalanceListener {
 
-    @Autowired
-    @Qualifier("kafkaProducer2.handler")
-    private lateinit var kafkaProducer2: KafkaProducerMessageHandler<*, *>
+				override fun onPartitionsRevoked(partitions: Collection<TopicPartition>) {
+					// empty
+				}
 
-    @Autowired
-    private lateinit var errorChannel: PollableChannel
+				override fun onPartitionsAssigned(partitions: Collection<TopicPartition>) {
+					this@ContextConfiguration.replyContainerLatch.countDown()
+				}
 
-    @Autowired(required = false)
-    @Qualifier("topic1ListenerContainer")
-    private lateinit var messageListenerContainer: MessageListenerContainer
+			}
+			return KafkaMessageListenerContainer(consumerFactory(), containerProperties)
+		}
 
-    @Autowired(required = false)
-    @Qualifier("kafkaTemplate:test-topic1")
-    private lateinit var kafkaTemplateTopic1: KafkaTemplate<Any, Any>
+		@Bean
+		fun serverGateway() =
+				IntegrationFlows.from(
+						Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory()))
+						.transform<String, String> { it.toUpperCase() }
+						.get()
 
-    @Autowired(required = false)
-    @Qualifier("kafkaTemplate:test-topic2")
-    private lateinit var kafkaTemplateTopic2: KafkaTemplate<*, *>
+		private fun containerProperties() =
+				ContainerProperties(TEST_TOPIC4)
+						.also {
+							it.groupId = "inGateGroup"
+						}
 
-    @Autowired
-    private lateinit var mapper: DefaultKafkaHeaderMapper
+	}
 
-    @Autowired
-    private lateinit var config: ContextConfiguration
+	interface Gate {
 
-    @Autowired
-    private lateinit var gate: Gate
+		fun exchange(@Header(KafkaHeaders.TOPIC) topic: String, out: String): String
 
-    @Test
-    fun testKafkaAdapters() {
-
-        this.sendToKafkaFlowInput.send(GenericMessage("foo", hashMapOf<String, Any>("foo" to "bar")))
-
-        assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper)
-
-        for (i in 0..99) {
-            val receive = this.listeningFromKafkaResults1.receive(20000)
-            assertThat(receive).isNotNull()
-            assertThat(receive!!.payload).isEqualTo("FOO")
-            val headers = receive.headers
-            assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue()
-            val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java)
-            acknowledgment?.acknowledge()
-            assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC1)
-            assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1)
-            assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0)
-            assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong())
-            assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME")
-            assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048633L)
-            assertThat(headers["foo"]).isEqualTo("bar")
-        }
-
-        for (i in 0..99) {
-            val receive = this.listeningFromKafkaResults2.receive(20000)
-            assertThat(receive).isNotNull()
-            assertThat(receive!!.payload).isEqualTo("FOO")
-            val headers = receive.headers
-            assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue()
-            val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java)
-            acknowledgment?.acknowledge()
-            assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC2)
-            assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1)
-            assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0)
-            assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong())
-            assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME")
-            assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048644L)
-        }
-
-        val message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build()
-
-        this.sendToKafkaFlowInput.send(message)
-
-        assertThat(this.listeningFromKafkaResults1.receive(10)).isNull()
-
-        val error = this.errorChannel.receive(10000)
-
-        assertThat(error).isNotNull().isInstanceOf(ErrorMessage::class.java)
-
-        val payload = error?.payload
-
-        assertThat(payload).isNotNull().isInstanceOf(MessageRejectedException::class.java)
-
-        assertThat(this.messageListenerContainer).isNotNull()
-        assertThat(this.kafkaTemplateTopic1).isNotNull()
-        assertThat(this.kafkaTemplateTopic2).isNotNull()
-
-        this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo")
-        assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue()
-        assertThat(this.config.fromSource).isEqualTo("foo")
-    }
-
-    @Test
-    fun testGateways() {
-        assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS))
-        assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO")
-    }
-
-    @Configuration
-    @EnableIntegration
-    @EnableKafka
-    class ContextConfiguration {
-
-        val sourceFlowLatch = CountDownLatch(1)
-
-        val replyContainerLatch = CountDownLatch(1)
-
-        var fromSource: Any? = null
-
-        @Bean
-        fun consumerFactory(): ConsumerFactory<Int, String> {
-            val props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka.embeddedKafka)
-            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
-            return DefaultKafkaConsumerFactory(props)
-        }
-
-        @Bean
-        fun errorChannel() = QueueChannel()
-
-        @Bean
-        fun topic1ListenerFromKafkaFlow() =
-                IntegrationFlows.from(
-                        Kafka.messageDrivenChannelAdapter<Int, String>(consumerFactory(),
-                                KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
-                                .configureListenerContainer({ c ->
-                                    c.ackMode(ContainerProperties.AckMode.MANUAL)
-                                            .id("topic1ListenerContainer")
-                                })
-                                .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(),
-                                        RawRecordHeaderErrorMessageStrategy()))
-                                .retryTemplate(RetryTemplate())
-                                .filterInRetry(true))
-                        .filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 },
-                                { f -> f.throwExceptionOnRejection(true) })
-                        .transform { it: String -> it.toUpperCase() }
-                        .channel { c -> c.queue("listeningFromKafkaResults1") }
-                        .get()
-
-        @Bean
-        fun topic2ListenerFromKafkaFlow() =
-                IntegrationFlows.from(
-                        Kafka.messageDrivenChannelAdapter<Int, String>(consumerFactory(),
-                                KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2)
-                                .configureListenerContainer({ c -> c.ackMode(ContainerProperties.AckMode.MANUAL) })
-                                .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(),
-                                        RawRecordHeaderErrorMessageStrategy()))
-                                .retryTemplate(RetryTemplate())
-                                .filterInRetry(true))
-                        .filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 },
-                                { f -> f.throwExceptionOnRejection(true) })
-                        .transform { it: String -> it.toUpperCase() }
-                        .channel { c -> c.queue("listeningFromKafkaResults2") }
-                        .get()
-
-        @Bean
-        fun producerFactory(): DefaultKafkaProducerFactory<Int, String> {
-            val props = KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka)
-            props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000")
-            return DefaultKafkaProducerFactory(props)
-        }
-
-        @Bean
-        fun sendToKafkaFlow() =
-                IntegrationFlow { f ->
-                    f.split<String>({ p -> Stream.generate { p }.limit(101) }, null)
-                            .publishSubscribeChannel { c ->
-                                c.subscribe { sf ->
-                                    sf.handle(
-                                            kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
-                                                    .timestampExpression("T(Long).valueOf('1487694048633')"),
-                                            { e -> e.id("kafkaProducer1") })
-                                }
-                                        .subscribe { sf ->
-                                            sf.handle(
-                                                    kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
-                                                            .timestamp<Any> { _ -> 1487694048644L },
-                                                    { e -> e.id("kafkaProducer2") })
-                                        }
-                            }
-                }
-
-        @Bean
-        fun mapper() = DefaultKafkaHeaderMapper()
-
-        private fun kafkaMessageHandler(producerFactory: ProducerFactory<Int, String>, topic: String) =
-                Kafka.outboundChannelAdapter(producerFactory)
-                        .messageKey<Any> { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] }
-                        .headerMapper(mapper())
-                        .sync(true)
-                        .partitionId<Any> { _ -> 0 }
-                        .topicExpression("headers[kafka_topic] ?: '$topic'")
-                        .configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") }
-
-
-        @Bean
-        fun sourceFlow() =
-                IntegrationFlows
-                        .from(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3))) { e -> e.poller(Pollers.fixedDelay(100)) }
-                        .handle({ p ->
-                            this.fromSource = p.getPayload()
-                            this.sourceFlowLatch.countDown()
-                        })
-                        .get()
-
-        @Bean
-        fun replyingKafkaTemplate() =
-                ReplyingKafkaTemplate(producerFactory(), replyContainer())
-                        .also {
-                            it.setDefaultReplyTimeout(Duration.ofSeconds(30))
-                        }
-
-        @Bean
-        fun outboundGateFlow() =
-                IntegrationFlows.from(Gate::class.java)
-                        .handle(Kafka.outboundGateway(replyingKafkaTemplate())
-                                .sync(true))
-                        .get()
-
-        private fun replyContainer(): GenericMessageListenerContainer<Int, String> {
-            val containerProperties = ContainerProperties(TEST_TOPIC5)
-            containerProperties.setGroupId("outGate")
-            containerProperties.setConsumerRebalanceListener(object : ConsumerRebalanceListener {
-
-                override fun onPartitionsRevoked(partitions: Collection<TopicPartition>) {
-                    // empty
-                }
-
-                override fun onPartitionsAssigned(partitions: Collection<TopicPartition>) {
-                    this@ContextConfiguration.replyContainerLatch.countDown()
-                }
-
-            })
-            return KafkaMessageListenerContainer(consumerFactory(), containerProperties)
-        }
-
-        @Bean
-        fun serverGateway() =
-                IntegrationFlows.from(
-                        Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory()))
-                        .transform { it: String -> it.toUpperCase() }
-                        .get()
-
-        private fun containerProperties() =
-                ContainerProperties(TEST_TOPIC4)
-                        .also {
-                            it.setGroupId("inGateGroup")
-                        }
-
-    }
-
-    interface Gate {
-
-        fun exchange(@Header(KafkaHeaders.TOPIC) topic: String, out: String): String
-
-    }
+	}
 
 }