Kafka binder tests migration to JUnit 5

2022-03-11 15:07:37 -05:00
parent 1c4f5e24dc
commit e51ade49db
14 changed files with 193 additions and 171 deletions
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AdminConfigTests.java
@@ -61,7 +61,7 @@ public class AdminConfigTests {
 	private KafkaMessageChannelBinder binder;

 	@Test
-	public void testConsumerTopicProperties() {
+	void testConsumerTopicProperties() {
 		final KafkaConsumerProperties consumerProperties = this.binder
 				.getExtendedConsumerProperties("secondInput");
 		final KafkaTopicProperties kafkaTopicProperties = consumerProperties.getTopic();
@@ -74,7 +74,7 @@ public class AdminConfigTests {
 	}

 	@Test
-	public void testProducerTopicProperties() {
+	void testProducerTopicProperties() {
 		final KafkaProducerProperties producerProperties = this.binder
 				.getExtendedProducerProperties("output");
 		final KafkaTopicProperties kafkaTopicProperties = producerProperties.getTopic();
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AutoCreateTopicDisabledTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/AutoCreateTopicDisabledTests.java
@@ -47,7 +47,7 @@ public class AutoCreateTopicDisabledTests {
 	private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();

 	@Test
-	public void testAutoCreateTopicDisabledFailsOnConsumerIfTopicNonExistentOnBroker() {
+	void testAutoCreateTopicDisabledFailsOnConsumerIfTopicNonExistentOnBroker() {

 		KafkaProperties kafkaProperties = new TestKafkaProperties();
 		kafkaProperties.setBootstrapServers(Collections
@@ -75,7 +75,7 @@ public class AutoCreateTopicDisabledTests {
 	}

 	@Test
-	public void testAutoCreateTopicDisabledFailsOnProducerIfTopicNonExistentOnBroker() {
+	void testAutoCreateTopicDisabledFailsOnProducerIfTopicNonExistentOnBroker() {

 		KafkaProperties kafkaProperties = new TestKafkaProperties();
 		kafkaProperties.setBootstrapServers(Collections
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderAutoConfigurationPropertiesTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -24,8 +24,8 @@ import java.util.Map;

 import org.apache.kafka.common.serialization.LongDeserializer;
 import org.apache.kafka.common.serialization.LongSerializer;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;

 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -38,7 +38,7 @@ import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.test.context.TestPropertySource;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.util.ReflectionUtils;

 import static org.assertj.core.api.Assertions.assertThat;
@@ -46,7 +46,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Ilayaperumal Gopinathan
 */
-@RunWith(SpringJUnit4ClassRunner.class)
+@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = { KafkaBinderConfiguration.class })
@TestPropertySource(locations = "classpath:binder-config-autoconfig.properties")
 public class KafkaBinderAutoConfigurationPropertiesTest {
@@ -59,7 +59,7 @@ public class KafkaBinderAutoConfigurationPropertiesTest {

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testKafkaBinderConfigurationWithKafkaProperties() throws Exception {
+	void testKafkaBinderConfigurationWithKafkaProperties() throws Exception {
 		assertThat(this.kafkaMessageChannelBinder).isNotNull();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = new ExtendedProducerProperties<>(
 				new KafkaProducerProperties());
@@ -113,7 +113,7 @@ public class KafkaBinderAutoConfigurationPropertiesTest {

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testKafkaHealthIndicatorProperties() {
+	void testKafkaHealthIndicatorProperties() {
 		assertThat(this.kafkaBinderHealthIndicator).isNotNull();
 		Field consumerFactoryField = ReflectionUtils.findField(
 				KafkaBinderHealthIndicator.class, "consumerFactory",
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationPropertiesTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -25,8 +25,8 @@ import java.util.Map;

 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
 import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;

 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
@@ -39,7 +39,7 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerPro
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.test.context.TestPropertySource;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.util.ReflectionUtils;

 import static org.assertj.core.api.Assertions.assertThat;
@@ -47,7 +47,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Ilayaperumal Gopinathan
 */
-@RunWith(SpringJUnit4ClassRunner.class)
+@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaAutoConfiguration.class,
 		KafkaBinderConfigurationPropertiesTest.class })
@TestPropertySource(locations = "classpath:binder-config.properties")
@@ -58,7 +58,7 @@ public class KafkaBinderConfigurationPropertiesTest {

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testKafkaBinderConfigurationProperties() throws Exception {
+	void testKafkaBinderConfigurationProperties() throws Exception {
 		assertThat(this.kafkaMessageChannelBinder).isNotNull();
 		KafkaProducerProperties kafkaProducerProperties = new KafkaProducerProperties();
 		kafkaProducerProperties.setBufferSize(12345);
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderConfigurationTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-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.
@@ -18,15 +18,15 @@ package org.springframework.cloud.stream.binder.kafka;

 import java.lang.reflect.Field;

-import org.junit.Test;
-import org.junit.runner.RunWith;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;

 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
 import org.springframework.kafka.support.ProducerListener;
-import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
 import org.springframework.util.ReflectionUtils;

 import static org.assertj.core.api.Assertions.assertThat;
@@ -34,7 +34,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 /**
 * @author Ilayaperumal Gopinathan
 */
-@RunWith(SpringJUnit4ClassRunner.class)
+@ExtendWith(SpringExtension.class)
@SpringBootTest(classes = { KafkaBinderConfiguration.class, KafkaAutoConfiguration.class,
 		KafkaBinderConfigurationTest.class })
 public class KafkaBinderConfigurationTest {
@@ -43,7 +43,7 @@ public class KafkaBinderConfigurationTest {
 	private KafkaMessageChannelBinder kafkaMessageChannelBinder;

 	@Test
-	public void testKafkaBinderProducerListener() {
+	void testKafkaBinderProducerListener() {
 		assertThat(this.kafkaMessageChannelBinder).isNotNull();
 		Field producerListenerField = ReflectionUtils.findField(
 				KafkaMessageChannelBinder.class, "producerListener",
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderHealthIndicatorTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2021 the original author or authors.
+ * Copyright 2017-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.
@@ -27,8 +27,10 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
@@ -49,6 +51,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 * @author Chukwubuikem Ume-Ugwa
 * @author Taras Danylchuk
 */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
 public class KafkaBinderHealthIndicatorTest {

 	private static final String TEST_TOPIC = "test";
@@ -74,7 +77,7 @@ public class KafkaBinderHealthIndicatorTest {

 	private final Map<String, KafkaMessageChannelBinder.TopicInformation> topicsInUse = new HashMap<>();

-	@Before
+	@BeforeEach
 	public void setup() {
 		MockitoAnnotations.initMocks(this);
 		org.mockito.BDDMockito.given(consumerFactory.createConsumer())
@@ -85,7 +88,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsUpWithNoConsumers() {
+	void kafkaBinderIsUpWithNoConsumers() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group1-healthIndicator", partitions, false));
@@ -100,7 +103,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsUp() {
+	void kafkaBinderIsUp() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group1-healthIndicator", partitions, false));
@@ -119,7 +122,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsDownWhenOneOfConsumersIsNotRunning() {
+	void kafkaBinderIsDownWhenOneOfConsumersIsNotRunning() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group1-healthIndicator", partitions, false));
@@ -138,7 +141,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsDownWhenOneOfContainersWasStoppedAbnormally() {
+	void kafkaBinderIsDownWhenOneOfContainersWasStoppedAbnormally() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group1-healthIndicator", partitions, false));
@@ -167,7 +170,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsUpWithRegexTopic() {
+	void kafkaBinderIsUpWithRegexTopic() {
 		topicsInUse.put(REGEX_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"regex-healthIndicator", null, true));
 		Health health = indicator.health();
@@ -179,7 +182,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsDown() {
+	void kafkaBinderIsDown() {
 		final List<PartitionInfo> partitions = partitions(null);
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group2-healthIndicator", partitions, false));
@@ -190,7 +193,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsDownWhenConsiderDownWhenAnyPartitionHasNoLeaderIsTrue() {
+	void kafkaBinderIsDownWhenConsiderDownWhenAnyPartitionHasNoLeaderIsTrue() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		partitions.add(new PartitionInfo(TEST_TOPIC, 0, null, null, null));
 		indicator.setConsiderDownWhenAnyPartitionHasNoLeader(true);
@@ -203,7 +206,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void kafkaBinderIsUpWhenConsiderDownWhenAnyPartitionHasNoLeaderIsFalse() {
+	void kafkaBinderIsUpWhenConsiderDownWhenAnyPartitionHasNoLeaderIsFalse() {
 		Node node = new Node(0, null, 0);
 		final List<PartitionInfo> partitions = partitions(node);
 		partitions.add(new PartitionInfo(TEST_TOPIC, 0, null, null, null));
@@ -216,8 +219,9 @@ public class KafkaBinderHealthIndicatorTest {
 		assertThat(health.getStatus()).isEqualTo(Status.UP);
 	}

-	@Test(timeout = 5000)
-	public void kafkaBinderDoesNotAnswer() {
+	@Test
+	@Timeout(5)
+	void kafkaBinderDoesNotAnswer() {
 		final List<PartitionInfo> partitions = partitions(new Node(-1, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group3-healthIndicator", partitions, false));
@@ -233,7 +237,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void createsConsumerOnceWhenInvokedMultipleTimes() {
+	void createsConsumerOnceWhenInvokedMultipleTimes() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"group4-healthIndicator", partitions, false));
@@ -248,7 +252,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void consumerCreationFailsFirstTime() {
+	void consumerCreationFailsFirstTime() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new KafkaMessageChannelBinder.TopicInformation(
 				"foo-healthIndicator", partitions, false));
@@ -267,7 +271,7 @@ public class KafkaBinderHealthIndicatorTest {
 	}

 	@Test
-	public void testIfNoTopicsRegisteredByTheBinderProvidesDownStatus() {
+	void testIfNoTopicsRegisteredByTheBinderProvidesDownStatus() {
 		Health health = indicator.health();
 		assertThat(health.getStatus()).isEqualTo(Status.UNKNOWN);
 	}
@@ -277,5 +281,4 @@ public class KafkaBinderHealthIndicatorTest {
 		partitions.add(new PartitionInfo(TEST_TOPIC, 0, leader, null, null));
 		return partitions;
 	}
-
 }
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-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.
@@ -20,17 +20,18 @@ import javax.security.auth.login.AppConfigurationEntry;

 import com.sun.security.auth.login.ConfigFile;
 import org.apache.kafka.common.security.JaasUtils;
-import org.junit.AfterClass;
-import org.junit.Before;
-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.BeforeEach;
+import org.junit.jupiter.api.Test;

 import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfiguration;
 import org.springframework.core.io.ClassPathResource;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
+import org.springframework.kafka.test.context.EmbeddedKafka;

 import static org.assertj.core.api.Assertions.assertThat;

@@ -38,28 +39,28 @@ import static org.assertj.core.api.Assertions.assertThat;
 * @author Marius Bogoevici
 * @author Soby Chacko
 */
+@EmbeddedKafka
 public class KafkaBinderJaasInitializerListenerTest {

 	private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers";

-	@ClassRule
-	public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true);
+	private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker();

 	private static String JAVA_LOGIN_CONFIG_PARAM_VALUE;

 	private final ApplicationContextRunner contextRunner = new ApplicationContextRunner()
 			.withUserConfiguration(KafkaBinderConfiguration.class, KafkaAutoConfiguration.class);

-	@BeforeClass
+	@BeforeAll
 	public static void setup() {
 		System.setProperty(KAFKA_BROKERS_PROPERTY,
-				kafkaEmbedded.getEmbeddedKafka().getBrokersAsString());
+				embeddedKafka.getBrokersAsString());
 		//Retrieve the current value for this system property if there is one set.
 		JAVA_LOGIN_CONFIG_PARAM_VALUE = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
 		System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
 	}

-	@AfterClass
+	@AfterAll
 	public static void clean() {
 		System.clearProperty(KAFKA_BROKERS_PROPERTY);
 		//If there was a previous value for this property, then restore it.
@@ -68,13 +69,13 @@ public class KafkaBinderJaasInitializerListenerTest {
 		}
 	}

-	@Before
+	@BeforeEach
 	public void before() {
 		System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
 	}

 	@Test
-	public void testConfigurationParsedCorrectlyWithKafkaClientAndDefaultControlFlag()
+	void testConfigurationParsedCorrectlyWithKafkaClientAndDefaultControlFlag()
 			throws Exception {
 		ConfigFile configFile = new ConfigFile(
 				new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
@@ -103,7 +104,7 @@ public class KafkaBinderJaasInitializerListenerTest {
 	}

 	@Test
-	public void testConfigurationParsedCorrectlyWithKafkaClientAndNonDefaultControlFlag()
+	void testConfigurationParsedCorrectlyWithKafkaClientAndNonDefaultControlFlag()
 			throws Exception {
 		ConfigFile configFile = new ConfigFile(
 				new ClassPathResource("jaas-sample-kafka-only.conf").getURI());
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-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.
@@ -31,8 +31,8 @@ import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
 import org.mockito.Mockito;
@@ -72,7 +72,7 @@ public class KafkaBinderMetricsTest {
 	@Mock
 	private KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties;

-	@Before
+	@BeforeEach
 	public void setup() {
 		MockitoAnnotations.openMocks(this);
 		org.mockito.BDDMockito.given(consumerFactory
@@ -88,7 +88,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void shouldIndicateLag() {
+	void shouldIndicateLag() {
 		final Map<TopicPartition, OffsetAndMetadata> committed = new HashMap<>();
 		TopicPartition topicPartition = new TopicPartition(TEST_TOPIC, 0);
 		committed.put(topicPartition, new OffsetAndMetadata(500));
@@ -108,7 +108,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void shouldNotContainAnyMetricsWhenUsingNoopGauge() {
+	void shouldNotContainAnyMetricsWhenUsingNoopGauge() {
 		// Adding NoopGauge for the offset metric.
 		meterRegistry.config().meterFilter(
 				MeterFilter.denyNameStartsWith("spring.cloud.stream.binder.kafka.offset"));
@@ -129,7 +129,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void shouldSumUpPartitionsLags() {
+	void shouldSumUpPartitionsLags() {
 		Map<TopicPartition, Long> endOffsets = new HashMap<>();
 		endOffsets.put(new TopicPartition(TEST_TOPIC, 0), 1000L);
 		endOffsets.put(new TopicPartition(TEST_TOPIC, 1), 1000L);
@@ -158,7 +158,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void shouldIndicateFullLagForNotCommittedGroups() {
+	void shouldIndicateFullLagForNotCommittedGroups() {
 		List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC,
 				new TopicInformation("group3-metrics", partitions, false));
@@ -172,7 +172,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void shouldNotCalculateLagForProducerTopics() {
+	void shouldNotCalculateLagForProducerTopics() {
 		List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC, new TopicInformation(null, partitions, false));
 		metrics.bindTo(meterRegistry);
@@ -180,7 +180,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void createsConsumerOnceWhenInvokedMultipleTimes() {
+	void createsConsumerOnceWhenInvokedMultipleTimes() {
 		final List<PartitionInfo> partitions = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC,
 				new TopicInformation("group4-metrics", partitions, false));
@@ -197,7 +197,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void consumerCreationFailsFirstTime() {
+	void consumerCreationFailsFirstTime() {
 		org.mockito.BDDMockito
 				.given(consumerFactory.createConsumer(ArgumentMatchers.any(),
 						ArgumentMatchers.any()))
@@ -219,7 +219,7 @@ public class KafkaBinderMetricsTest {
 	}

 	@Test
-	public void createOneConsumerPerGroup() {
+	void createOneConsumerPerGroup() {
 		final List<PartitionInfo> partitions1 = partitions(new Node(0, null, 0));
 		final List<PartitionInfo> partitions2 = partitions(new Node(0, null, 0));
 		topicsInUse.put(TEST_TOPIC,
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
@@ -343,7 +343,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testDefaultHeaderMapper() throws Exception {
+	void testDefaultHeaderMapper() throws Exception {
 		Binder binder = getBinder();

 		BindingProperties producerBindingProperties = createProducerBindingProperties(
@@ -402,7 +402,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testCustomHeaderMapper() throws Exception {
+	void testCustomHeaderMapper() throws Exception {

 		KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
 		binderConfiguration.setHeaderMapperBeanName("headerMapper");
@@ -480,7 +480,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testWellKnownHeaderMapperWithBeanNameKafkaHeaderMapper() throws Exception {
+	void testWellKnownHeaderMapperWithBeanNameKafkaHeaderMapper() throws Exception {

 		KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();

@@ -673,7 +673,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testSendAndReceiveBatch() throws Exception {
+	void testSendAndReceiveBatch() throws Exception {
 		Binder binder = getBinder();
 		BindingProperties outputBindingProperties = createProducerBindingProperties(
 				createProducerProperties());
@@ -735,7 +735,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testDlqWithNativeSerializationEnabledOnDlqProducer() throws Exception {
+	void testDlqWithNativeSerializationEnabledOnDlqProducer() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();

@@ -815,7 +815,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testDlqWithNativeDecodingOnConsumerButMissingSerializerOnDlqProducer()
+	void testDlqWithNativeDecodingOnConsumerButMissingSerializerOnDlqProducer()
 			throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
@@ -889,7 +889,7 @@ public class KafkaBinderTests extends
 	// For more details on the context of this test: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/657
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testDlqWithProducerPropertiesSetAtBinderLevel()
+	void testDlqWithProducerPropertiesSetAtBinderLevel()
 			throws Exception {

 		KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
@@ -956,37 +956,37 @@ public class KafkaBinderTests extends
 	}

 	@Test
-	public void testDlqAndRetry() throws Exception {
+	void testDlqAndRetry() throws Exception {
 		testDlqGuts(true, null, null, false, false);
 	}

 	@Test
-	public void testDlqAndRetryTransactional() throws Exception {
+	void testDlqAndRetryTransactional() throws Exception {
 		testDlqGuts(true, null, null, true, false);
 	}

 	@Test
-	public void testDlq() throws Exception {
+	void testDlq() throws Exception {
 		testDlqGuts(false, null, 3, false, false);
 	}

 	@Test
-	public void testDlqWithDlqDestinationResolver() throws Exception {
+	void testDlqWithDlqDestinationResolver() throws Exception {
 		testDlqGuts(false, null, 3, false, true);
 	}

 	@Test
-	public void testDlqTransactional() throws Exception {
+	void testDlqTransactional() throws Exception {
 		testDlqGuts(false, null, 3, true, false);
 	}

 	@Test
-	public void testDlqNone() throws Exception {
+	void testDlqNone() throws Exception {
 		testDlqGuts(false, HeaderMode.none, 1, false, false);
 	}

 	@Test
-	public void testDlqEmbedded() throws Exception {
+	void testDlqEmbedded() throws Exception {
 		testDlqGuts(false, HeaderMode.embeddedHeaders, 3, false, false);
 	}

@@ -1227,7 +1227,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testDefaultAutoCommitOnErrorWithDlq() throws Exception {
+	void testDefaultAutoCommitOnErrorWithDlq() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		BindingProperties producerBindingProperties = createProducerBindingProperties(
@@ -1304,7 +1304,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testRetriesWithoutDlq() throws Exception {
+	void testRetriesWithoutDlq() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		BindingProperties producerBindingProperties = createProducerBindingProperties(
@@ -1349,7 +1349,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testCommonErrorHandlerBeanNameOnConsumerBinding() throws Exception {
+	void testCommonErrorHandlerBeanNameOnConsumerBinding() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		BindingProperties producerBindingProperties = createProducerBindingProperties(
@@ -1412,7 +1412,7 @@ public class KafkaBinderTests extends
 	//See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/870 for motivation for this test.
 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoCommitOnErrorWhenManualAcknowledgement() throws Exception {
+	void testAutoCommitOnErrorWhenManualAcknowledgement() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		BindingProperties producerBindingProperties = createProducerBindingProperties(
@@ -1492,7 +1492,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testConfigurableDlqName() throws Exception {
+	void testConfigurableDlqName() throws Exception {
 		Binder binder = getBinder();

 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
@@ -1586,7 +1586,7 @@ public class KafkaBinderTests extends
 	// TODO: This test needs to be rethought - sending byte[] without explicit content
 	// type
 	// - yet being converted by the json converter
-	public void testCompression() throws Exception {
+	void testCompression() throws Exception {
 		final KafkaProducerProperties.CompressionType[] codecs = new KafkaProducerProperties.CompressionType[] {
 				KafkaProducerProperties.CompressionType.none,
 				KafkaProducerProperties.CompressionType.gzip,
@@ -1639,7 +1639,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testEarliest() throws Exception {
+	void testEarliest() throws Exception {
 		Binding<MessageChannel> producerBinding = null;
 		Binding<MessageChannel> consumerBinding = null;

@@ -1787,7 +1787,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff() throws Exception {
+	void testManualAckSucceedsWhenAutoCommitOffsetIsTurnedOff() throws Exception {
 		Binder binder = getBinder();

 		DirectChannel moduleOutputChannel = createBindableChannel("output",
@@ -1833,7 +1833,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testManualAckIsNotPossibleWhenAutoCommitOffsetIsEnabledOnTheBinder()
+	void testManualAckIsNotPossibleWhenAutoCommitOffsetIsEnabledOnTheBinder()
 			throws Exception {
 		Binder binder = getBinder();

@@ -2052,7 +2052,7 @@ public class KafkaBinderTests extends
 	@Test
 //	@Override
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testPartitionedModuleJava() throws Exception {
+	void testPartitionedModuleJava() throws Exception {
 		Binder binder = getBinder();

 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
@@ -2209,7 +2209,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testPartitionedModuleJavaWithRawMode() throws Exception {
+	void testPartitionedModuleJavaWithRawMode() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setHeaderMode(HeaderMode.none);
@@ -2272,7 +2272,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testPartitionedModuleSpELWithRawMode() throws Exception {
+	void testPartitionedModuleSpELWithRawMode() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setPartitionKeyExpression(
@@ -2344,7 +2344,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testPartitionedNative() throws Exception {
+	void testPartitionedNative() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> properties = createProducerProperties();
 		properties.setPartitionCount(6);
@@ -2377,7 +2377,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testSendAndReceiveWithRawMode() throws Exception {
+	void testSendAndReceiveWithRawMode() throws Exception {
 		Binder binder = getBinder();

 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
@@ -2422,7 +2422,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testAllowNonTransactionalProducerSetting() throws Exception {
+	void testAllowNonTransactionalProducerSetting() throws Exception {
 		AbstractKafkaTestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = createBindableChannel("output",
 				new BindingProperties());
@@ -2444,7 +2444,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testProducerErrorChannel() throws Exception {
+	void testProducerErrorChannel() throws Exception {
 		AbstractKafkaTestBinder binder = getBinder();
 		DirectChannel moduleOutputChannel = createBindableChannel("output",
 				new BindingProperties());
@@ -2510,7 +2510,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoCreateTopicsEnabledSucceeds() throws Exception {
+	void testAutoCreateTopicsEnabledSucceeds() throws Exception {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
 		configurationProperties.setAutoCreateTopics(true);
 		Binder binder = getBinder(configurationProperties);
@@ -2526,7 +2526,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testCustomPartitionCountOverridesDefaultIfLarger() throws Exception {
+	void testCustomPartitionCountOverridesDefaultIfLarger() throws Exception {
 		byte[] testPayload = new byte[2048];
 		Arrays.fill(testPayload, (byte) 65);
 		KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
@@ -2563,7 +2563,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testCustomPartitionCountDoesNotOverridePartitioningIfSmaller()
+	void testCustomPartitionCountDoesNotOverridePartitioningIfSmaller()
 			throws Exception {
 		byte[] testPayload = new byte[2048];
 		Arrays.fill(testPayload, (byte) 65);
@@ -2601,7 +2601,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testDynamicKeyExpression() throws Exception {
+	void testDynamicKeyExpression() throws Exception {
 		Binder binder = getBinder(createConfigurationProperties());
 		QueueChannel moduleInputChannel = new QueueChannel();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
@@ -2635,7 +2635,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception {
+	void testCustomPartitionCountOverridesPartitioningIfLarger() throws Exception {
 		byte[] testPayload = new byte[2048];
 		Arrays.fill(testPayload, (byte) 65);
 		KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
@@ -2671,7 +2671,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testDefaultConsumerStartsAtEarliest() throws Exception {
+	void testDefaultConsumerStartsAtEarliest() throws Exception {
 		Binder binder = getBinder(createConfigurationProperties());

 		BindingProperties producerBindingProperties = createProducerBindingProperties(
@@ -2737,7 +2737,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testResume() throws Exception {
+	void testResume() throws Exception {
 		Binding<MessageChannel> producerBinding = null;
 		Binding<MessageChannel> consumerBinding = null;

@@ -2832,7 +2832,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testSyncProducerMetadata() throws Exception {
+	void testSyncProducerMetadata() throws Exception {
 		Binder binder = getBinder(createConfigurationProperties());
 		DirectChannel output = new DirectChannel();
 		String testTopicName = UUID.randomUUID().toString();
@@ -2852,7 +2852,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testSendTimeoutExpressionProducerMetadata() throws Exception {
+	void testSendTimeoutExpressionProducerMetadata() throws Exception {
 		Binder binder = getBinder(createConfigurationProperties());
 		DirectChannel output = new DirectChannel();
 		String testTopicName = UUID.randomUUID().toString();
@@ -2874,7 +2874,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoCreateTopicsDisabledOnBinderStillWorksAsLongAsBrokerCreatesTopic()
+	void testAutoCreateTopicsDisabledOnBinderStillWorksAsLongAsBrokerCreatesTopic()
 			throws Exception {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
 		configurationProperties.setAutoCreateTopics(false);
@@ -2921,7 +2921,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting()
+	void testAutoConfigureTopicsDisabledSucceedsIfTopicExisting()
 			throws Throwable {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

@@ -2941,7 +2941,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Throwable {
+	void testPartitionCountIncreasedIfAutoAddPartitionsSet() throws Throwable {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

 		String testTopicName = "existing" + System.currentTimeMillis();
@@ -2960,7 +2960,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled()
+	void testAutoAddPartitionsDisabledSucceedsIfTopicUnderPartitionedAndAutoRebalanceEnabled()
 			throws Throwable {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

@@ -2987,7 +2987,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled()
+	void testAutoAddPartitionsDisabledFailsIfTopicUnderPartitionedAndAutoRebalanceDisabled()
 			throws Throwable {
 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();

@@ -3016,7 +3016,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly()
+	void testAutoAddPartitionsDisabledSucceedsIfTopicPartitionedCorrectly()
 			throws Throwable {
 		Binding<?> binding = null;
 		try {
@@ -3060,7 +3060,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testPartitionCountNotReduced() throws Throwable {
+	void testPartitionCountNotReduced() throws Throwable {
 		String testTopicName = "existing" + System.currentTimeMillis();

 		KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
@@ -3084,7 +3084,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testConsumerDefaultDeserializer() throws Throwable {
+	void testConsumerDefaultDeserializer() throws Throwable {
 		Binding<?> binding = null;
 		try {
 			KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
@@ -3116,7 +3116,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testConsumerCustomDeserializer() throws Exception {
+	void testConsumerCustomDeserializer() throws Exception {
 		Binding<?> binding = null;
 		try {
 			KafkaBinderConfigurationProperties configurationProperties = createConfigurationProperties();
@@ -3155,7 +3155,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testNativeSerializationWithCustomSerializerDeserializer()
+	void testNativeSerializationWithCustomSerializerDeserializer()
 			throws Exception {
 		Binding<?> producerBinding = null;
 		Binding<?> consumerBinding = null;
@@ -3218,7 +3218,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testNativeSerializationWithCustomSerializerDeserializerBytesPayload()
+	void testNativeSerializationWithCustomSerializerDeserializerBytesPayload()
 			throws Exception {
 		Binding<?> producerBinding = null;
 		Binding<?> consumerBinding = null;
@@ -3277,7 +3277,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testBuiltinSerialization() throws Exception {
+	void testBuiltinSerialization() throws Exception {
 		Binding<?> producerBinding = null;
 		Binding<?> consumerBinding = null;
 		try {
@@ -3342,7 +3342,7 @@ public class KafkaBinderTests extends
 	 */
 	@Test
 	@SuppressWarnings({ "unchecked", "rawtypes" })
-	public void testSendAndReceiveWithMixedMode() throws Exception {
+	void testSendAndReceiveWithMixedMode() throws Exception {
 		KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
 		binderConfiguration.setHeaders("foo");
 		Binder binder = getBinder(binderConfiguration);
@@ -3440,7 +3440,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testPolledConsumer() throws Exception {
+	void testPolledConsumer() throws Exception {
 		KafkaTestBinder binder = getBinder();
 		PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(
 				this.messageConverter);
@@ -3494,7 +3494,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testPolledConsumerRequeue() throws Exception {
+	void testPolledConsumerRequeue() throws Exception {
 		KafkaTestBinder binder = getBinder();
 		PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(
 				this.messageConverter);
@@ -3528,7 +3528,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testPolledConsumerWithDlq() throws Exception {
+	void testPolledConsumerWithDlq() throws Exception {
 		KafkaTestBinder binder = getBinder();
 		PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(
 				this.messageConverter);
@@ -3573,7 +3573,7 @@ public class KafkaBinderTests extends

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testTopicPatterns() throws Exception {
+	void testTopicPatterns() throws Exception {
 		try (AdminClient admin = AdminClient.create(
 				Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
 						embeddedKafka.getBrokersAsString()))) {
@@ -3606,7 +3606,7 @@ public class KafkaBinderTests extends
 	}

 	@Test
-	public void testSameTopicCannotBeProvisionedAgain() throws Throwable {
+	void testSameTopicCannotBeProvisionedAgain() throws Throwable {
 		CountDownLatch latch = new CountDownLatch(1);
 		try (AdminClient admin = AdminClient.create(
 				Collections.singletonMap(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
@@ -3630,7 +3630,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testResetOffsets() throws Exception {
+	void testResetOffsets() throws Exception {
 		Binding<?> producerBinding = null;
 		Binding<?> consumerBinding = null;
 		try {
@@ -3704,7 +3704,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testRecordMetadata() throws Exception {
+	void testRecordMetadata() throws Exception {
 		Binding<?> producerBinding = null;
 		try {
 			String testPayload = "test";
@@ -3741,7 +3741,7 @@ public class KafkaBinderTests extends

 	@Test
 	@SuppressWarnings("unchecked")
-	public void testMessageKeyInPayload() throws Exception {
+	void testMessageKeyInPayload() throws Exception {
 		Binding<?> producerBinding = null;
 		try {
 			String testPayload = "test";
@@ -3778,17 +3778,17 @@ public class KafkaBinderTests extends
 	}

 	@Test
-	public void testInternalHeadersNotPropagated() throws Exception {
+	void testInternalHeadersNotPropagated() throws Exception {
 		testInternalHeadersNotPropagatedGuts("propagate.1", null, null);
 	}

 	@Test
-	public void testInternalHeadersNotPropagatedCustomHeader() throws Exception {
+	void testInternalHeadersNotPropagatedCustomHeader() throws Exception {
 		testInternalHeadersNotPropagatedGuts("propagate.2", new String[] { "foo", "*" }, null);
 	}

 	@Test
-	public void testInternalHeadersNotPropagatedCustomMapper() throws Exception {
+	void testInternalHeadersNotPropagatedCustomMapper() throws Exception {
 		testInternalHeadersNotPropagatedGuts("propagate.3", null, new BinderHeaderMapper("*"));
 	}

@@ -3870,7 +3870,7 @@ public class KafkaBinderTests extends
 	}

 	@Test
-	public void testNoBrokerOverride() throws Exception {
+	void testNoBrokerOverride() throws Exception {
 		Binder binder = getBinder();
 		ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
 		producerProperties.getExtension().getConfiguration().put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "foo");
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -36,7 +36,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.serialization.Deserializer;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;

 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -73,7 +73,7 @@ import static org.mockito.Mockito.verify;
 public class KafkaBinderUnitTests {

 	@Test
-	public void testPropertyOverrides() throws Exception {
+	void testPropertyOverrides() throws Exception {
 		KafkaProperties kafkaProperties = new TestKafkaProperties();
 		KafkaBinderConfigurationProperties binderConfigurationProperties = new KafkaBinderConfigurationProperties(
 				kafkaProperties);
@@ -119,7 +119,7 @@ public class KafkaBinderUnitTests {
 	}

 	@Test
-	public void testMergedConsumerProperties() {
+	void testMergedConsumerProperties() {
 		KafkaProperties bootProps = new TestKafkaProperties();
 		bootProps.getConsumer().getProperties()
 				.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "bar");
@@ -136,7 +136,7 @@ public class KafkaBinderUnitTests {
 	}

 	@Test
-	public void testMergedProducerProperties() {
+	void testMergedProducerProperties() {
 		KafkaProperties bootProps = new TestKafkaProperties();
 		bootProps.getProducer().getProperties().put(ProducerConfig.RETRIES_CONFIG, "bar");
 		KafkaBinderConfigurationProperties props = new KafkaBinderConfigurationProperties(
@@ -152,30 +152,30 @@ public class KafkaBinderUnitTests {
 	}

 	@Test
-	public void testOffsetResetWithGroupManagementEarliest() throws Exception {
+	void testOffsetResetWithGroupManagementEarliest() throws Exception {
 		testOffsetResetWithGroupManagement(true, true, "foo-100",
 				"testOffsetResetWithGroupManagementEarliest");
 	}

 	@Test
-	public void testOffsetResetWithGroupManagementLatest() throws Throwable {
+	void testOffsetResetWithGroupManagementLatest() throws Throwable {
 		testOffsetResetWithGroupManagement(false, true, "foo-101",
 				"testOffsetResetWithGroupManagementLatest");
 	}

 	@Test
-	public void testOffsetResetWithManualAssignmentEarliest() throws Exception {
+	void testOffsetResetWithManualAssignmentEarliest() throws Exception {
 		testOffsetResetWithGroupManagement(true, false, "foo-102",
 				"testOffsetResetWithManualAssignmentEarliest");
 	}

 	@Test
-	public void testOffsetResetWithGroupManualAssignmentLatest() throws Throwable {
+	void testOffsetResetWithGroupManualAssignmentLatest() throws Throwable {
 		testOffsetResetWithGroupManagement(false, false, "foo-103",
 				"testOffsetResetWithGroupManualAssignmentLatest");
 	}

-	private void testOffsetResetWithGroupManagement(final boolean earliest,
+	void testOffsetResetWithGroupManagement(final boolean earliest,
 			boolean groupManage, String topic, String group) throws Exception {

 		final List<TopicPartition> partitions = new ArrayList<>();
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2022 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
@@ -23,8 +23,8 @@ import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.TopicPartition;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 import org.mockito.InOrder;

 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
@@ -36,7 +36,9 @@ import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
+import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.util.concurrent.SettableListenableFuture;
@@ -55,19 +57,23 @@ import static org.mockito.Mockito.spy;
 * @since 2.0
 *
 */
+@EmbeddedKafka(count = 1, controlledShutdown = true, brokerProperties = {"transaction.state.log.replication.factor=1",
+	"transaction.state.log.min.isr=1"})
 public class KafkaTransactionTests {

-	@ClassRule
-	public static final EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1)
-			.brokerProperty("transaction.state.log.replication.factor", "1")
-			.brokerProperty("transaction.state.log.min.isr", "1");
+	private static EmbeddedKafkaBroker embeddedKafka;
+
+	@BeforeAll
+	public static void setup() {
+		embeddedKafka = EmbeddedKafkaCondition.getBroker();
+	}

 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
-	public void testProducerRunsInTx() {
+	void testProducerRunsInTx() {
 		KafkaProperties kafkaProperties = new TestKafkaProperties();
 		kafkaProperties.setBootstrapServers(Collections
-				.singletonList(embeddedKafka.getEmbeddedKafka().getBrokersAsString()));
+				.singletonList(embeddedKafka.getBrokersAsString()));
 		KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(
 				kafkaProperties);
 		configurationProperties.getTransaction().setTransactionIdPrefix("foo-");
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderBootstrapTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderBootstrapTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-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.
@@ -16,31 +16,36 @@

 package org.springframework.cloud.stream.binder.kafka.bootstrap;

-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;

 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
+import org.springframework.kafka.test.context.EmbeddedKafka;

 /**
 * @author Marius Bogoevici
 */
+@EmbeddedKafka(count = 1, controlledShutdown = true)
 public class KafkaBinderBootstrapTest {

-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10);
+	private static EmbeddedKafkaBroker embeddedKafka;
+
+	@BeforeAll
+	public static void setup() {
+		embeddedKafka = EmbeddedKafkaCondition.getBroker();
+	}

 	@Test
-	public void testKafkaBinderConfiguration() throws Exception {
+	void testKafkaBinderConfiguration() throws Exception {
 		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(
 				SimpleApplication.class).web(WebApplicationType.NONE).run(
 						"--spring.cloud.stream.kafka.binder.brokers="
-								+ embeddedKafka.getEmbeddedKafka().getBrokersAsString(),
-						"--spring.cloud.stream.kafka.binder.zkNodes=" + embeddedKafka
-								.getEmbeddedKafka().getZookeeperConnectionString());
+								+ embeddedKafka.getBrokersAsString());
 		applicationContext.close();
 	}

--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderCustomHealthCheckTests.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderCustomHealthCheckTests.java
@@ -16,8 +16,8 @@

 package org.springframework.cloud.stream.binder.kafka.bootstrap;

-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;

 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.boot.WebApplicationType;
@@ -28,7 +28,9 @@ import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealth;
 import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
-import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
+import org.springframework.kafka.test.EmbeddedKafkaBroker;
+import org.springframework.kafka.test.condition.EmbeddedKafkaCondition;
+import org.springframework.kafka.test.context.EmbeddedKafka;

 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
@@ -36,17 +38,22 @@ import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy;
 /**
 * @author Soby Chacko
 */
+@EmbeddedKafka(count = 1, controlledShutdown = true)
 public class KafkaBinderCustomHealthCheckTests {

-	@ClassRule
-	public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10);
+	private static EmbeddedKafkaBroker embeddedKafka;
+
+	@BeforeAll
+	public static void setup() {
+		embeddedKafka = EmbeddedKafkaCondition.getBroker();
+	}

 	@Test
-	public void testCustomHealthIndicatorIsActivated() {
+	void testCustomHealthIndicatorIsActivated() {
 		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(
 				CustomHealthCheckApplication.class).web(WebApplicationType.NONE).run(
 				"--spring.cloud.stream.kafka.binder.brokers="
-						+ embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+						+ embeddedKafka.getBrokersAsString());
 		final KafkaBinderHealth kafkaBinderHealth = applicationContext.getBean(KafkaBinderHealth.class);
 		assertThat(kafkaBinderHealth).isInstanceOf(CustomHealthIndicator.class);
 		assertThatThrownBy(() -> applicationContext.getBean(KafkaBinderHealthIndicator.class)).isInstanceOf(NoSuchBeanDefinitionException.class);
--- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java
+++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java
@@ -71,7 +71,7 @@ public class KafkaBinderMeterRegistryTest {
 	}

 	@Test
-	public void testMetricsWithSingleBinder() throws Exception {
+	void testMetricsWithSingleBinder() throws Exception {
 		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class)
 				.web(WebApplicationType.NONE)
 				.run("--spring.cloud.stream.bindings.uppercase-in-0.destination=inputTopic",
@@ -99,7 +99,7 @@ public class KafkaBinderMeterRegistryTest {
 	}

 	@Test
-	public void testMetricsWithMultiBinders() {
+	void testMetricsWithMultiBinders() {
 		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(SimpleApplication.class)
 				.web(WebApplicationType.NONE)
 				.run("--spring.cloud.stream.bindings.uppercase-in-0.destination=inputTopic",