diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml index 5b8088848f..3d4075a5fc 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests-context.xml @@ -3,13 +3,9 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka" xmlns:context="http://www.springframework.org/schema/context" - xmlns:task="http://www.springframework.org/schema/task" - xmlns:util="http://www.springframework.org/schema/util" - xsi:schemaLocation="http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd - http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd + xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd - http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd - http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd"> + http://www.springframework.org/schema/integration/kafka http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd"> diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 8f0c4f45dc..0a15a2a607 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -66,7 +66,7 @@ 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.KafkaEmbedded; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -102,8 +102,8 @@ public class KafkaDslTests { private static final String TEST_TOPIC5 = "test-topic5"; @ClassRule - public static KafkaEmbedded embeddedKafka = - new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5); + public static EmbeddedKafkaRule embeddedKafka = + new EmbeddedKafkaRule(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5); @Autowired @Qualifier("sendToKafkaFlow.input") @@ -231,7 +231,8 @@ public class KafkaDslTests { @Bean public ConsumerFactory consumerFactory() { - Map props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka); + Map props = KafkaTestUtils + .consumerProps("test1", "false", embeddedKafka.getEmbeddedKafka()); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return new DefaultKafkaConsumerFactory<>(props); } @@ -276,8 +277,9 @@ public class KafkaDslTests { @Bean public IntegrationFlow topic2ListenerFromKafkaFlow() { return IntegrationFlows - .from(Kafka.messageDrivenChannelAdapter(kafkaListenerContainerFactory().createContainer(TEST_TOPIC2), - KafkaMessageDrivenChannelAdapter.ListenerMode.record) + .from(Kafka + .messageDrivenChannelAdapter(kafkaListenerContainerFactory().createContainer(TEST_TOPIC2), + KafkaMessageDrivenChannelAdapter.ListenerMode.record) .filterInRetry(true)) .filter(Message.class, m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101, @@ -289,7 +291,7 @@ public class KafkaDslTests { @Bean public ProducerFactory producerFactory() { - return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka)); + return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka())); } @Bean diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index e95c5fa419..cea31b7834 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -48,7 +48,8 @@ import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.MessagingMessageConverter; -import org.springframework.kafka.test.rule.KafkaEmbedded; +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; @@ -63,6 +64,7 @@ import org.springframework.retry.support.RetryTemplate; /** * * @author Gary Russell + * @author Artem Bilan * * @since 3.0.2 * @@ -82,8 +84,8 @@ public class InboundGatewayTests { private static String topic6 = "testTopic6"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, - topic6); + public static EmbeddedKafkaRule embeddedKafka = + new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); @Rule public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace().categories("org.apache.kafka.clients", @@ -91,7 +93,9 @@ public class InboundGatewayTests { @Test public void testInbound() throws Exception { - Map consumerProps = KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka); + EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); + Map consumerProps = + KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); Consumer consumer = cf2.createConsumer(); @@ -99,7 +103,7 @@ public class InboundGatewayTests { Map props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -156,6 +160,7 @@ public class InboundGatewayTests { @Test public void testInboundErrorRecover() throws Exception { + EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); Map consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); @@ -164,7 +169,7 @@ public class InboundGatewayTests { Map props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic3); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -231,6 +236,7 @@ public class InboundGatewayTests { @Test public void testInboundRetryErrorRecover() throws Exception { + EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); Map consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); @@ -239,7 +245,7 @@ public class InboundGatewayTests { Map props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic5); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 84bae87ebd..bf4e93ca34 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -73,7 +73,8 @@ import org.springframework.kafka.support.converter.ConversionException; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.kafka.support.converter.StringJsonMessageConverter; -import org.springframework.kafka.test.rule.KafkaEmbedded; +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; @@ -106,13 +107,16 @@ public class MessageDrivenAdapterTests { private static String topic5 = "testTopic5"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5); + public static EmbeddedKafkaRule embeddedKafkaRule = + new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); @Test public void testInboundRecord() throws Exception { Map props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -197,7 +201,7 @@ public class MessageDrivenAdapterTests { public void testInboundRecordRetryRecover() throws Exception { Map props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic4); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -224,7 +228,7 @@ public class MessageDrivenAdapterTests { ContainerTestUtils.waitForAssignment(container, 2); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic4); template.sendDefault(1, "foo"); @@ -249,7 +253,7 @@ public class MessageDrivenAdapterTests { public void testInboundRecordNoRetryRecover() throws Exception { Map props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic5); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -272,7 +276,7 @@ public class MessageDrivenAdapterTests { ContainerTestUtils.waitForAssignment(container, 2); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic5); template.sendDefault(1, "foo"); @@ -296,7 +300,7 @@ public class MessageDrivenAdapterTests { public void testInboundBatch() throws Exception { Map props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic2); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -319,7 +323,7 @@ public class MessageDrivenAdapterTests { ContainerTestUtils.waitForAssignment(container, 2); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic2); template.sendDefault(0, 1487694048607L, 1, "foo"); @@ -373,7 +377,7 @@ public class MessageDrivenAdapterTests { public void testInboundJson() throws Exception { Map props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(props); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic3); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); @@ -386,7 +390,7 @@ public class MessageDrivenAdapterTests { ContainerTestUtils.waitForAssignment(container, 2); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic3); Headers kHeaders = new RecordHeaders(); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java index c2e4cd5833..d3818d3db4 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceIntegrationTests.java @@ -32,12 +32,15 @@ import org.junit.Test; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.test.rule.KafkaEmbedded; +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; /** * @author Gary Russell + * @author Artem Bilan + * * @since 3.0.1 * */ @@ -46,7 +49,9 @@ public class MessageSourceIntegrationTests { public static final String TOPIC1 = "MessageSourceIntegrationTests1"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, 1, TOPIC1); + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, 1, TOPIC1); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); @Test public void testSource() throws Exception { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 8612923a6d..43863a5087 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -68,7 +68,8 @@ import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaNull; import org.springframework.kafka.support.SendResult; -import org.springframework.kafka.test.rule.KafkaEmbedded; +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; import org.springframework.messaging.MessageHandlingException; @@ -82,6 +83,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; /** * @author Gary Russell * @author Biju Kunjummen + * @author Artem Bilan * * @since 2.0 */ @@ -100,8 +102,10 @@ public class KafkaProducerMessageHandlerTests { private static String topic6 = "testTopic6in"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, - topic6); + public static EmbeddedKafkaRule embeddedKafkaRule = + new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); + + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); private static Consumer consumer; diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt index 32e034a500..4abb133a80 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/KafkaDslKotlinTests.kt @@ -17,13 +17,7 @@ package org.springframework.integration.kafka.dsl import assertk.assert -import assertk.assertions.contains -import assertk.assertions.isEqualTo -import assertk.assertions.isInstanceOf -import assertk.assertions.isNotNull -import assertk.assertions.isNull -import assertk.assertions.isSameAs -import assertk.assertions.isTrue +import assertk.assertions.* import assertk.catch import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.clients.consumer.ConsumerRebalanceListener @@ -50,11 +44,7 @@ import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessage import org.springframework.integration.support.MessageBuilder import org.springframework.integration.test.util.TestUtils import org.springframework.kafka.annotation.EnableKafka -import org.springframework.kafka.core.ConsumerFactory -import org.springframework.kafka.core.DefaultKafkaConsumerFactory -import org.springframework.kafka.core.DefaultKafkaProducerFactory -import org.springframework.kafka.core.KafkaTemplate -import org.springframework.kafka.core.ProducerFactory +import org.springframework.kafka.core.* import org.springframework.kafka.listener.ContainerProperties import org.springframework.kafka.listener.GenericMessageListenerContainer import org.springframework.kafka.listener.KafkaMessageListenerContainer @@ -63,7 +53,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.rule.KafkaEmbedded +import org.springframework.kafka.test.rule.EmbeddedKafkaRule import org.springframework.kafka.test.utils.KafkaTestUtils import org.springframework.messaging.Message import org.springframework.messaging.MessageChannel @@ -103,7 +93,7 @@ class KafkaDslKotlinTests { @ClassRule @JvmField - var embeddedKafka = KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5) + var embeddedKafka = EmbeddedKafkaRule(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5) } @@ -242,7 +232,7 @@ class KafkaDslKotlinTests { @Bean fun consumerFactory(): ConsumerFactory { - val props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka) + val props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka.embeddedKafka) props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") return DefaultKafkaConsumerFactory(props) } @@ -286,7 +276,7 @@ class KafkaDslKotlinTests { .get() @Bean - fun producerFactory() = DefaultKafkaProducerFactory(KafkaTestUtils.producerProps(embeddedKafka)) + fun producerFactory() = DefaultKafkaProducerFactory(KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka)) @Bean fun sendToKafkaFlow() =