From 690743ee0ce6e044516a25c899d4199b7be8db56 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 29 Jul 2022 12:17:12 -0400 Subject: [PATCH] Use Global Embedded Kafka whenever possible For better test suite lifecycle (higher performance) reuse one global embedded Kafka broker introduced in Spring for Apache Kafka `3.0` Some tests have left with their own `EmbeddedKafkaBroker` definitions since they rely on different partitions --- .../kafka/channnel/ChannelTests.java | 13 +- .../kafka/config/xml/AllXmlTests-context.xml | 10 +- .../kafka/config/xml/AllXmlTests.java | 2 - .../kafka/config/xml/ChannelParserTests.java | 27 +-- .../integration/kafka/dsl/KafkaDslTests.java | 33 ++-- .../inbound/MessageDrivenAdapterTests.java | 73 ++++--- .../MessageSourceIntegrationTests.java | 10 +- .../kafka/dsl/kotlin/KafkaDslKotlinTests.kt | 181 +++++++++--------- .../test/resources/junit-platform.properties | 3 + 9 files changed, 176 insertions(+), 176 deletions(-) create mode 100644 spring-integration-kafka/src/test/resources/junit-platform.properties diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java index c11dc706a0..e902e41151 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/channnel/ChannelTests.java @@ -27,6 +27,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.channel.NullChannel; @@ -44,8 +45,6 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.support.KafkaHeaders; -import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -61,7 +60,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; * */ @SpringJUnitConfig -@EmbeddedKafka(topics = { "channel.1", "channel.2", "channel.3" }, partitions = 1) public class ChannelTests { @Test @@ -118,17 +116,18 @@ public class ChannelTests { @Configuration public static class Config { - @Autowired - private EmbeddedKafkaBroker broker; + @Value("${spring.global.embedded.kafka.brokers}") + String embeddedKafkaBrokers; @Bean public ProducerFactory pf() { - return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(this.broker)); + return new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(this.embeddedKafkaBrokers)); } @Bean public ConsumerFactory cf() { - Map consumerProps = KafkaTestUtils.consumerProps("channelTests", "false", this.broker); + Map consumerProps = + KafkaTestUtils.consumerProps(this.embeddedKafkaBrokers, "channelTests", "false"); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return new DefaultKafkaConsumerFactory<>(consumerProps); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml index 581f84cee6..0adbc813e6 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests-context.xml @@ -3,14 +3,18 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration" xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka" + xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/integration/kafka https://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd - http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd"> + http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd + http://www.springframework.org/schema/context https://www.springframework.org/schema/context/spring-context.xsd"> + + - + @@ -22,7 +26,7 @@ - + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java index c6866a6c5b..6e5c511ae4 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/AllXmlTests.java @@ -22,7 +22,6 @@ import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.test.annotation.DirtiesContext; @@ -36,7 +35,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; */ @SpringJUnitConfig @DirtiesContext -@EmbeddedKafka(topics = { "one", "two", "three", "four" }) public class AllXmlTests { @Autowired diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java index 5e5c63357e..39840ad77f 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/ChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020 the original author or authors. + * Copyright 2020-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,16 +28,17 @@ import org.springframework.context.annotation.ImportResource; import org.springframework.integration.kafka.channel.PollableKafkaChannel; import org.springframework.integration.kafka.channel.SubscribableKafkaChannel; import org.springframework.integration.kafka.inbound.KafkaMessageSource; +import org.springframework.integration.test.util.TestUtils; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaOperations; import org.springframework.kafka.listener.ConsumerProperties; -import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell + * @author Artem Bilan * * @since 5.4 * @@ -65,17 +66,17 @@ public class ChannelParserTests { @Test void testParser() { - assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "topic")).isEqualTo("ptpTopic"); - assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "topic")).isEqualTo("pubSubTopic"); - assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "container")).isNotNull(); - assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "container")).isNotNull(); - assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "template")).isSameAs(this.template); - assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "template")).isSameAs(this.template); - assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "template")).isSameAs(this.template); - assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "source")).isSameAs(this.source); - assertThat(KafkaTestUtils.getPropertyValue(this.ptp, "groupId")).isEqualTo("ptpGroup"); - assertThat(KafkaTestUtils.getPropertyValue(this.pubSub, "groupId")).isEqualTo("pubSubGroup"); - assertThat(KafkaTestUtils.getPropertyValue(this.pollable, "groupId")).isEqualTo("pollableGroup"); + assertThat(TestUtils.getPropertyValue(this.ptp, "topic")).isEqualTo("ptpTopic"); + assertThat(TestUtils.getPropertyValue(this.pubSub, "topic")).isEqualTo("pubSubTopic"); + assertThat(TestUtils.getPropertyValue(this.ptp, "container")).isNotNull(); + assertThat(TestUtils.getPropertyValue(this.pubSub, "container")).isNotNull(); + assertThat(TestUtils.getPropertyValue(this.ptp, "template")).isSameAs(this.template); + assertThat(TestUtils.getPropertyValue(this.pubSub, "template")).isSameAs(this.template); + assertThat(TestUtils.getPropertyValue(this.pollable, "template")).isSameAs(this.template); + assertThat(TestUtils.getPropertyValue(this.pollable, "source")).isSameAs(this.source); + assertThat(TestUtils.getPropertyValue(this.ptp, "groupId")).isEqualTo("ptpGroup"); + assertThat(TestUtils.getPropertyValue(this.pubSub, "groupId")).isEqualTo("pubSubGroup"); + assertThat(TestUtils.getPropertyValue(this.pollable, "groupId")).isEqualTo("pollableGroup"); } @Configuration 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 0a7d95ddfe..d66eeec64d 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 @@ -35,6 +35,7 @@ import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -71,8 +72,6 @@ import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; -import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -97,9 +96,6 @@ import org.springframework.util.backoff.FixedBackOff; */ @SpringJUnitConfig @DirtiesContext -@EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3, - KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5, KafkaDslTests.TEST_TOPIC6, KafkaDslTests.TEST_TOPIC7, - KafkaDslTests.TEST_TOPIC8, KafkaDslTests.TEST_TOPIC9 }) public class KafkaDslTests { static final String TEST_TOPIC1 = "test-topic1"; @@ -118,8 +114,6 @@ public class KafkaDslTests { static final String TEST_TOPIC8 = "test-topic8"; - static final String TEST_TOPIC9 = "test-topic9"; - @Autowired @Qualifier("sendToKafkaFlow.input") private MessageChannel sendToKafkaFlowInput; @@ -266,7 +260,8 @@ public class KafkaDslTests { assertThat(received) .isNotNull() .extracting("payload") - .isEqualTo("foo"); } + .isEqualTo("foo"); + } @Configuration @EnableIntegration @@ -281,14 +276,12 @@ public class KafkaDslTests { private Object fromSource; - @Autowired - private EmbeddedKafkaBroker embeddedKafka; - + @Value("${spring.global.embedded.kafka.brokers}") + String embeddedKafkaBrokers; @Bean public ConsumerFactory consumerFactory() { - Map props = KafkaTestUtils - .consumerProps("test1", "false", this.embeddedKafka); + Map props = KafkaTestUtils.consumerProps(this.embeddedKafkaBrokers, "test1", "false"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return new DefaultKafkaConsumerFactory<>(props); } @@ -302,7 +295,7 @@ public class KafkaDslTests { public IntegrationFlow topic1ListenerFromKafkaFlow() { return IntegrationFlow .from(Kafka.messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) .configureListenerContainer(c -> c.ackMode(ContainerProperties.AckMode.MANUAL) .idleEventInterval(100L) @@ -344,7 +337,7 @@ public class KafkaDslTests { @Bean public ProducerFactory producerFactory() { - Map props = KafkaTestUtils.producerProps(this.embeddedKafka); + Map props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers); props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000); return new DefaultKafkaProducerFactory<>(props); } @@ -479,8 +472,8 @@ public class KafkaDslTests { public IntegrationFlow serverGateway() { return IntegrationFlow .from(Kafka.inboundGateway(consumerFactory(), containerProperties(), - producerFactory()) - .configureListenerContainer(container -> container.errorHandler(eh()))) + producerFactory()) + .configureListenerContainer(container -> container.errorHandler(eh()))) .transform(String::toUpperCase) .get(); } @@ -495,9 +488,9 @@ public class KafkaDslTests { ContainerProperties props = containerProperties(); props.setGroupId("wreh"); return IntegrationFlow.from(Kafka.messageDrivenChannelAdapter(consumerFactory(), props) - .configureListenerContainer(container -> { - container.errorHandler(recoveringErrorHandler()); - })) + .configureListenerContainer(container -> { + container.errorHandler(recoveringErrorHandler()); + })) .handle(p -> { throw new RuntimeException("test"); }) 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 e06a01e7a0..604a3369f2 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 @@ -50,6 +50,7 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -85,8 +86,6 @@ import org.springframework.kafka.support.converter.ConversionException; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.kafka.support.converter.StringJsonMessageConverter; -import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.ContainerTestUtils; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; @@ -111,13 +110,6 @@ import org.springframework.retry.support.RetryTemplate; * @since 5.4 * */ -@EmbeddedKafka(controlledShutdown = true, - topics = { MessageDrivenAdapterTests.topic1, - MessageDrivenAdapterTests.topic2, - MessageDrivenAdapterTests.topic3, - MessageDrivenAdapterTests.topic4, - MessageDrivenAdapterTests.topic5, - MessageDrivenAdapterTests.topic6 }) class MessageDrivenAdapterTests { static final String topic1 = "testTopic1"; @@ -132,9 +124,16 @@ class MessageDrivenAdapterTests { static final String topic6 = "testTopic6"; + static String EMBEDDED_BROKERS; + + @BeforeAll + static void setup() { + EMBEDDED_BROKERS = System.getProperty("spring.global.embedded.kafka.brokers"); + } + @Test - void testInboundRecord(EmbeddedKafkaBroker embeddedKafka) { - Map props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka); + void testInboundRecord() { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test1", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); @@ -155,9 +154,9 @@ class MessageDrivenAdapterTests { }); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic1); @@ -219,8 +218,8 @@ class MessageDrivenAdapterTests { } @Test - void testInboundRecordRetryRecover(EmbeddedKafkaBroker embeddedKafka) { - Map props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka); + void testInboundRecordRetryRecover() { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test4", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic4); @@ -248,9 +247,9 @@ class MessageDrivenAdapterTests { adapter.setRetryTemplate(retryTemplate); adapter.afterPropertiesSet(); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic4); @@ -286,8 +285,8 @@ class MessageDrivenAdapterTests { * to the consumer. */ @Test - void testInboundRecordRetryRecoverWithoutRecoveryCallback(EmbeddedKafkaBroker embeddedKafka) throws Exception { - Map props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka); + void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test6", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic6); @@ -321,9 +320,9 @@ class MessageDrivenAdapterTests { adapter.afterPropertiesSet(); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic6); @@ -336,8 +335,8 @@ class MessageDrivenAdapterTests { } @Test - void testInboundRecordNoRetryRecover(EmbeddedKafkaBroker embeddedKafka) { - Map props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka); + void testInboundRecordNoRetryRecover() { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test5", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic5); @@ -362,9 +361,9 @@ class MessageDrivenAdapterTests { adapter.setBindSourceRecord(true); adapter.afterPropertiesSet(); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic5); @@ -391,8 +390,8 @@ class MessageDrivenAdapterTests { } @Test - void testInboundBatch(EmbeddedKafkaBroker embeddedKafka) throws Exception { - Map props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka); + void testInboundBatch() throws Exception { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test2", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic2); @@ -421,9 +420,9 @@ class MessageDrivenAdapterTests { }); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic2); @@ -479,8 +478,8 @@ class MessageDrivenAdapterTests { } @Test - void testInboundJson(EmbeddedKafkaBroker embeddedKafka) { - Map props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka); + void testInboundJson() { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test3", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic3); @@ -492,9 +491,9 @@ class MessageDrivenAdapterTests { adapter.setOutputChannel(out); adapter.afterPropertiesSet(); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic3); @@ -521,8 +520,8 @@ class MessageDrivenAdapterTests { } @Test - void testInboundJsonWithPayload(EmbeddedKafkaBroker embeddedKafka) { - Map props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka); + void testInboundJsonWithPayload() { + Map props = KafkaTestUtils.consumerProps(EMBEDDED_BROKERS, "test6", "true"); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic6); @@ -538,9 +537,9 @@ class MessageDrivenAdapterTests { adapter.setOutputChannel(out); adapter.afterPropertiesSet(); adapter.start(); - ContainerTestUtils.waitForAssignment(container, 2); + ContainerTestUtils.waitForAssignment(container, 1); - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + Map senderProps = KafkaTestUtils.producerProps(EMBEDDED_BROKERS); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic6); 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 1df46cec3d..0360c1136b 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 @@ -34,8 +34,6 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.ConsumerProperties; -import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; @@ -48,14 +46,14 @@ import org.springframework.messaging.support.GenericMessage; * @since 5.4 * */ -@EmbeddedKafka(controlledShutdown = true, topics = MessageSourceIntegrationTests.TOPIC1, partitions = 1) class MessageSourceIntegrationTests { static final String TOPIC1 = "MessageSourceIntegrationTests1"; @Test - void testSource(EmbeddedKafkaBroker embeddedKafka) throws Exception { - Map consumerProps = KafkaTestUtils.consumerProps("testSource", "false", embeddedKafka); + void testSource() throws Exception { + String brokers = System.getProperty("spring.global.embedded.kafka.brokers"); + Map consumerProps = KafkaTestUtils.consumerProps(brokers, "testSource", "false"); consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); consumerProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 42); @@ -81,7 +79,7 @@ class MessageSourceIntegrationTests { KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, consumerProperties); - Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + Map producerProps = KafkaTestUtils.producerProps(brokers); DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(producerProps); KafkaTemplate template = new KafkaTemplate<>(producerFactory); template.setDefaultTopic(TOPIC1); diff --git a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt index ec65d3e946..0c1d546310 100644 --- a/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt +++ b/spring-integration-kafka/src/test/kotlin/org/springframework/integration/kafka/dsl/kotlin/KafkaDslKotlinTests.kt @@ -23,10 +23,9 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener import org.apache.kafka.clients.producer.ProducerConfig import org.apache.kafka.common.TopicPartition import org.junit.jupiter.api.Test -import org.junit.jupiter.api.condition.DisabledOnOs -import org.junit.jupiter.api.condition.OS import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier +import org.springframework.beans.factory.annotation.Value import org.springframework.context.annotation.Bean import org.springframework.context.annotation.Configuration import org.springframework.integration.IntegrationMessageHeaderAccessor @@ -48,8 +47,6 @@ import org.springframework.kafka.requestreply.ReplyingKafkaTemplate import org.springframework.kafka.support.Acknowledgment import org.springframework.kafka.support.DefaultKafkaHeaderMapper import org.springframework.kafka.support.KafkaHeaders -import org.springframework.kafka.test.EmbeddedKafkaBroker -import org.springframework.kafka.test.context.EmbeddedKafka import org.springframework.kafka.test.utils.KafkaTestUtils import org.springframework.messaging.Message import org.springframework.messaging.MessageChannel @@ -72,24 +69,21 @@ import java.util.stream.Stream * @since 5.4 */ -@DisabledOnOs(OS.WINDOWS) @SpringJUnitConfig @DirtiesContext -@EmbeddedKafka(topics = [KafkaDslKotlinTests.TEST_TOPIC1, KafkaDslKotlinTests.TEST_TOPIC2, - KafkaDslKotlinTests.TEST_TOPIC3, KafkaDslKotlinTests.TEST_TOPIC4, KafkaDslKotlinTests.TEST_TOPIC5]) class KafkaDslKotlinTests { companion object { - const val TEST_TOPIC1 = "test-topic1" + const val TEST_TOPIC1 = "test-kotlin-topic1" - const val TEST_TOPIC2 = "test-topic2" + const val TEST_TOPIC2 = "test-kotlin-topic2" - const val TEST_TOPIC3 = "test-topic3" + const val TEST_TOPIC3 = "test-kotlin-topic3" - const val TEST_TOPIC4 = "test-topic4" + const val TEST_TOPIC4 = "test-kotlin-topic4" - const val TEST_TOPIC5 = "test-topic5" + const val TEST_TOPIC5 = "test-kotlin-topic5" } @@ -108,10 +102,6 @@ class KafkaDslKotlinTests { @Qualifier("kafkaProducer1.handler") private lateinit var kafkaProducer1: KafkaProducerMessageHandler<*, *> - @Autowired - @Qualifier("kafkaProducer2.handler") - private lateinit var kafkaProducer2: KafkaProducerMessageHandler<*, *> - @Autowired private lateinit var errorChannel: PollableChannel @@ -120,11 +110,11 @@ class KafkaDslKotlinTests { private lateinit var messageListenerContainer: MessageListenerContainer @Autowired(required = false) - @Qualifier("kafkaTemplate:test-topic1") + @Qualifier("kafkaTemplate:test-kotlin-topic1") private lateinit var kafkaTemplateTopic1: KafkaTemplate @Autowired(required = false) - @Qualifier("kafkaTemplate:test-topic2") + @Qualifier("kafkaTemplate:test-kotlin-topic2") private lateinit var kafkaTemplateTopic2: KafkaTemplate<*, *> @Autowired @@ -215,12 +205,12 @@ class KafkaDslKotlinTests { var fromSource: Any? = null - @Autowired - private lateinit var embeddedKafka: EmbeddedKafkaBroker + @Value("\${spring.global.embedded.kafka.brokers}") + lateinit var embeddedKafkaBrokers: String @Bean fun consumerFactory(): ConsumerFactory { - val props = KafkaTestUtils.consumerProps("test1", "false", this.embeddedKafka) + val props = KafkaTestUtils.consumerProps(this.embeddedKafkaBrokers, "test1", "false") props[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest" return DefaultKafkaConsumerFactory(props) } @@ -230,96 +220,111 @@ class KafkaDslKotlinTests { @Bean fun topic1ListenerFromKafkaFlow() = - integrationFlow( - Kafka.messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) - .configureListenerContainer { - it.ackMode(ContainerProperties.AckMode.MANUAL) - .id("topic1ListenerContainer") - } - .errorChannel(errorChannel()) - .retryTemplate(RetryTemplate()) - .filterInRetry(true)) { - filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) { throwExceptionOnRejection(true) } - transform { it.uppercase() } - channel { queue("listeningFromKafkaResults1") } + integrationFlow( + Kafka.messageDrivenChannelAdapter( + consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1 + ) + .configureListenerContainer { + it.ackMode(ContainerProperties.AckMode.MANUAL) + .id("topic1ListenerContainer") + } + .errorChannel(errorChannel()) + .retryTemplate(RetryTemplate()) + .filterInRetry(true)) { + filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) { + throwExceptionOnRejection( + true + ) } + transform { it.uppercase() } + channel { queue("listeningFromKafkaResults1") } + } @Bean fun topic2ListenerFromKafkaFlow() = - integrationFlow( - Kafka.messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2) - .configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) } - .errorChannel(errorChannel()) - .retryTemplate(RetryTemplate()) - .filterInRetry(true)) { - filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) { throwExceptionOnRejection(true) } - transform { it.uppercase() } - channel { queue("listeningFromKafkaResults2") } + integrationFlow( + Kafka.messageDrivenChannelAdapter( + consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2 + ) + .configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) } + .errorChannel(errorChannel()) + .retryTemplate(RetryTemplate()) + .filterInRetry(true)) { + filter>({ m -> (m.headers[KafkaHeaders.RECEIVED_KEY] as Int) < 101 }) { + throwExceptionOnRejection( + true + ) } + transform { it.uppercase() } + channel { queue("listeningFromKafkaResults2") } + } @Bean fun producerFactory(): DefaultKafkaProducerFactory { - val props = KafkaTestUtils.producerProps(this.embeddedKafka) + val props = KafkaTestUtils.producerProps(this.embeddedKafkaBrokers) props[ProducerConfig.MAX_BLOCK_MS_CONFIG] = "10000" return DefaultKafkaProducerFactory(props) } @Bean fun sendToKafkaFlow() = - integrationFlow { - split { p -> Stream.generate { p }.limit(101) } - publishSubscribe(PublishSubscribeChannel(), - { - handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC1) - .timestampExpression("T(Long).valueOf('1487694048633')") - ) { id("kafkaProducer1") } - }, - { - handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2) - .timestamp { 1487694048644L } - ) { id("kafkaProducer2") } - } - ) - } + integrationFlow { + split { p -> Stream.generate { p }.limit(101) } + publishSubscribe(PublishSubscribeChannel(), + { + handle( + kafkaMessageHandler(producerFactory(), TEST_TOPIC1) + .timestampExpression("T(Long).valueOf('1487694048633')") + ) { id("kafkaProducer1") } + }, + { + handle(kafkaMessageHandler(producerFactory(), TEST_TOPIC2) + .timestamp { 1487694048644L } + ) { id("kafkaProducer2") } + } + ) + } @Bean fun mapper() = DefaultKafkaHeaderMapper() private fun kafkaMessageHandler(producerFactory: ProducerFactory, topic: String) = - Kafka.outboundChannelAdapter(producerFactory) - .messageKey { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } - .headerMapper(mapper()) - .sync(true) - .partitionId { 0 } - .topicExpression("headers[kafka_topic] ?: '$topic'") - .configureKafkaTemplate { it.id("kafkaTemplate:$topic") } + Kafka.outboundChannelAdapter(producerFactory) + .messageKey { it.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } + .headerMapper(mapper()) + .sync(true) + .partitionId { 0 } + .topicExpression("headers[kafka_topic] ?: '$topic'") + .configureKafkaTemplate { it.id("kafkaTemplate:$topic") } @Bean fun sourceFlow() = - integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)), - { poller(Pollers.fixedDelay(100)) }) { - handle { m -> - this@ContextConfiguration.fromSource = m.payload - this@ContextConfiguration.sourceFlowLatch.countDown() - } + integrationFlow(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3)), + { poller(Pollers.fixedDelay(100)) }) { + handle { m -> + this@ContextConfiguration.fromSource = m.payload + this@ContextConfiguration.sourceFlowLatch.countDown() } + } @Bean fun replyingKafkaTemplate() = - ReplyingKafkaTemplate(producerFactory(), replyContainer()) - .also { - it.setDefaultReplyTimeout(Duration.ofSeconds(30)) - } + ReplyingKafkaTemplate(producerFactory(), replyContainer()) + .also { + it.setDefaultReplyTimeout(Duration.ofSeconds(30)) + } @Bean fun outboundGateFlow() = - integrationFlow { - handle(Kafka.outboundGateway(replyingKafkaTemplate()) - .sync(true)) - } + integrationFlow { + handle( + Kafka.outboundGateway(replyingKafkaTemplate()) + .sync(true) + ) + } private fun replyContainer(): GenericMessageListenerContainer { val containerProperties = ContainerProperties(TEST_TOPIC5) @@ -340,15 +345,15 @@ class KafkaDslKotlinTests { @Bean fun serverGateway() = - integrationFlow(Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) { - transform { it.uppercase() } - } + integrationFlow(Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) { + transform { it.uppercase() } + } private fun containerProperties() = - ContainerProperties(TEST_TOPIC4) - .also { - it.setGroupId("inGateGroup") - } + ContainerProperties(TEST_TOPIC4) + .also { + it.setGroupId("inGateGroup") + } } diff --git a/spring-integration-kafka/src/test/resources/junit-platform.properties b/spring-integration-kafka/src/test/resources/junit-platform.properties new file mode 100644 index 0000000000..33083f2674 --- /dev/null +++ b/spring-integration-kafka/src/test/resources/junit-platform.properties @@ -0,0 +1,3 @@ +spring.kafka.global.embedded.enabled = true +spring.embedded.kafka.brokers.property=spring.global.embedded.kafka.brokers +spring.kafka.embedded.partitions=1