Fix KafkaProducerMessageHandlerTests for NPE on BeanFactory

* Migrate `KafkaDslTests` and `KafkaDslKotlinTests` to their own Embedded Kafka instead of global one.
Looks like there is some race condition when single Kafka broker is running for the whole test suite.

Author: artembilan

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

@@ -71,6 +71,7 @@
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
+import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -97,6 +98,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
+@EmbeddedKafka
 public class KafkaDslTests {
 
 	private static final Log log = LogFactory.getLog(KafkaDslTests.class);
@@ -275,7 +277,7 @@ public static class ContextConfiguration {
 
 		private Object fromSource;
 
-		@Value("${spring.global.embedded.kafka.brokers}")
+		@Value("${spring.kafka.bootstrap-servers}")
 		String embeddedKafkaBrokers;
 
 		@Bean

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

@@ -728,6 +728,7 @@ void testTxNonTx() {
 		given(template.getProducerFactory()).willReturn(mock(ProducerFactory.class));
 		KafkaProducerMessageHandler<String, String> handler = new KafkaProducerMessageHandler<>(template);
 		handler.setTopicExpression(new LiteralExpression("topic"));
+		handler.setBeanFactory(mock());
 		handler.handleMessage(new GenericMessage<>("foo"));
 		verify(template, never()).executeInTransaction(any());
 		verify(template).send(any(ProducerRecord.class));

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

@@ -47,6 +47,7 @@ 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.context.EmbeddedKafka
 import org.springframework.kafka.test.utils.KafkaTestUtils
 import org.springframework.messaging.Message
 import org.springframework.messaging.MessageChannel
@@ -71,6 +72,7 @@ import java.util.stream.Stream
 
 @SpringJUnitConfig
 @DirtiesContext
+@EmbeddedKafka
 class KafkaDslKotlinTests {
 
 	companion object {
@@ -205,7 +207,7 @@ class KafkaDslKotlinTests {
 
 		var fromSource: Any? = null
 
-		@Value("\${spring.global.embedded.kafka.brokers}")
+		@Value("\${spring.kafka.bootstrap-servers}")
 		lateinit var embeddedKafkaBrokers: String
 
 		@Bean