From bf0134cb52c2d7f993ac1fc69d515e21c228d6a9 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 1 Oct 2019 14:48:07 -0400 Subject: [PATCH] Upgrade to latest GAs * Move all the test to JUnit 5; remove redundant JUnit 4 dependencies --- ...essageDrivenChannelAdapterParserTests.java | 19 +- .../xml/KafkaOutboundAdapterParserTests.java | 47 +- .../integration/kafka/dsl/KafkaDslTests.java | 36 +- .../kafka/inbound/InboundGatewayTests.java | 64 +-- .../inbound/MessageDrivenAdapterTests.java | 60 +- .../MessageSourceIntegrationTests.java | 28 +- .../kafka/inbound/MessageSourceTests.java | 81 +-- .../KafkaProducerMessageHandlerTests.java | 95 ++- .../kafka/dsl/KafkaDslKotlinTests.kt | 543 +++++++++--------- 9 files changed, 486 insertions(+), 487 deletions(-) diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java index 028a9127e5..159b28ed17 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaMessageDrivenChannelAdapterParserTests.java @@ -21,8 +21,7 @@ import static org.mockito.Mockito.mock; import java.util.Collections; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.channel.NullChannel; @@ -40,16 +39,16 @@ import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetryTemplate; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Artem Bilan. * @author Gary Russell */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration -public class KafkaMessageDrivenChannelAdapterParserTests { +@SpringJUnitConfig +@DirtiesContext +class KafkaMessageDrivenChannelAdapterParserTests { @Autowired private NullChannel nullChannel; @@ -73,7 +72,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests { private RecoveryCallback recoveryCallback; @Test - public void testKafkaMessageDrivenChannelAdapterParser() { + void testKafkaMessageDrivenChannelAdapterParser() { assertThat(this.kafkaListener.isAutoStartup()).isFalse(); assertThat(this.kafkaListener.isRunning()).isFalse(); assertThat(this.kafkaListener.getPhase()).isEqualTo(100); @@ -94,7 +93,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests { } @Test - public void testKafkaBatchMessageDrivenChannelAdapterParser() { + void testKafkaBatchMessageDrivenChannelAdapterParser() { assertThat(this.kafkaBatchListener.isAutoStartup()).isFalse(); assertThat(this.kafkaBatchListener.isRunning()).isFalse(); assertThat(this.kafkaBatchListener.getPhase()).isEqualTo(100); @@ -110,7 +109,7 @@ public class KafkaMessageDrivenChannelAdapterParserTests { @Test @SuppressWarnings("unchecked") - public void testKafkaMessageDrivenChannelAdapterOptions() { + void testKafkaMessageDrivenChannelAdapterOptions() { DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(Collections.emptyMap()); ContainerProperties containerProps = new ContainerProperties("foo"); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index a2fbcbbe63..00bc903d20 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -17,7 +17,7 @@ package org.springframework.integration.kafka.config.xml; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.Assert.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; @@ -29,8 +29,7 @@ import java.util.concurrent.TimeoutException; import org.apache.kafka.clients.producer.MockProducer; import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.StringSerializer; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -45,8 +44,7 @@ import org.springframework.kafka.core.ProducerFactory; import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Soby Chacko @@ -57,18 +55,17 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * * @since 0.5 */ -@RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration +@SpringJUnitConfig @DirtiesContext -public class KafkaOutboundAdapterParserTests { +class KafkaOutboundAdapterParserTests { @Autowired private ApplicationContext appContext; @Test - public void testOutboundAdapterConfiguration() { + void testOutboundAdapterConfiguration() { KafkaProducerMessageHandler messageHandler - = this.appContext.getBean("kafkaOutboundChannelAdapter.handler", KafkaProducerMessageHandler.class); + = this.appContext.getBean("kafkaOutboundChannelAdapter.handler", KafkaProducerMessageHandler.class); assertThat(messageHandler).isNotNull(); assertThat(messageHandler.getOrder()).isEqualTo(3); assertThat(TestUtils.getPropertyValue(messageHandler, "topicExpression.literalValue")).isEqualTo("foo"); @@ -98,7 +95,7 @@ public class KafkaOutboundAdapterParserTests { } @Test - public void testSyncMode() { + void testSyncMode() { MockProducer mockProducer = new MockProducer(false, new IntegerSerializer(), new StringSerializer()) { @@ -128,28 +125,18 @@ public class KafkaOutboundAdapterParserTests { return null; }); - try { - handler.handleMessage(new GenericMessage<>("foo")); - fail("MessageHandlingException expected"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageHandlingException.class); - assertThat(e.getCause()).isExactlyInstanceOf(KafkaProducerException.class); - assertThat(e.getCause().getCause()).isInstanceOf(RuntimeException.class); - assertThat(e.getMessage()).contains("Async Producer Mock exception"); - } + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo"))) + .withCauseInstanceOf(KafkaProducerException.class) + .withRootCauseInstanceOf(RuntimeException.class) + .withMessageContaining("Async Producer Mock exception"); handler.setSendTimeout(1); - try { - handler.handleMessage(new GenericMessage<>("foo")); - fail("MessageTimeoutException expected"); - } - catch (Exception e) { - assertThat(e).isInstanceOf(MessageTimeoutException.class); - assertThat(e.getCause()).isExactlyInstanceOf(TimeoutException.class); - assertThat(e.getMessage()).contains("Timeout waiting for response from KafkaProducer"); - } + assertThatExceptionOfType(MessageTimeoutException.class) + .isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo"))) + .withCauseInstanceOf(TimeoutException.class) + .withMessageContaining("Timeout waiting for response from KafkaProducer"); } } 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 1a10315eba..ca16ef513e 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 @@ -30,9 +30,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.TopicPartition; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -67,7 +65,8 @@ import org.springframework.kafka.listener.MessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.DefaultKafkaHeaderMapper; import org.springframework.kafka.support.KafkaHeaders; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -78,7 +77,7 @@ import org.springframework.messaging.support.ErrorMessage; import org.springframework.messaging.support.GenericMessage; import org.springframework.retry.support.RetryTemplate; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Artem Bilan @@ -89,23 +88,21 @@ import org.springframework.test.context.junit4.SpringRunner; * * @since 3.0 */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext +@EmbeddedKafka(topics = { KafkaDslTests.TEST_TOPIC1, KafkaDslTests.TEST_TOPIC2, KafkaDslTests.TEST_TOPIC3, + KafkaDslTests.TEST_TOPIC4, KafkaDslTests.TEST_TOPIC5 }) public class KafkaDslTests { - private static final String TEST_TOPIC1 = "test-topic1"; + static final String TEST_TOPIC1 = "test-topic1"; - private static final String TEST_TOPIC2 = "test-topic2"; + static final String TEST_TOPIC2 = "test-topic2"; - private static final String TEST_TOPIC3 = "test-topic3"; + static final String TEST_TOPIC3 = "test-topic3"; - private static final String TEST_TOPIC4 = "test-topic4"; + static final String TEST_TOPIC4 = "test-topic4"; - private static final String TEST_TOPIC5 = "test-topic5"; - - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = - new EmbeddedKafkaRule(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5); + static final String TEST_TOPIC5 = "test-topic5"; @Autowired @Qualifier("sendToKafkaFlow.input") @@ -151,7 +148,6 @@ public class KafkaDslTests { @Test public void testKafkaAdapters() throws Exception { - this.sendToKafkaFlowInput.send(new GenericMessage<>("foo", Collections.singletonMap("foo", "bar"))); assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper); @@ -230,10 +226,14 @@ public class KafkaDslTests { private Object fromSource; + @Autowired + private EmbeddedKafkaBroker embeddedKafka; + + @Bean public ConsumerFactory consumerFactory() { Map props = KafkaTestUtils - .consumerProps("test1", "false", embeddedKafka.getEmbeddedKafka()); + .consumerProps("test1", "false", this.embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); return new DefaultKafkaConsumerFactory<>(props); } @@ -295,7 +295,7 @@ public class KafkaDslTests { @Bean public ProducerFactory producerFactory() { - Map props = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka()); + Map props = KafkaTestUtils.producerProps(this.embeddedKafka); props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000); return new DefaultKafkaProducerFactory<>(props); } 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 ff615a804f..44b47d9236 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 @@ -29,9 +29,9 @@ import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.junit.ClassRule; -import org.junit.Rule; -import org.junit.Test; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -40,7 +40,6 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.rule.Log4j2LevelAdjuster; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; @@ -52,7 +51,6 @@ import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.ContainerTestUtils; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; @@ -76,7 +74,7 @@ import org.springframework.retry.support.RetryTemplate; * @since 3.0.2 * */ -public class InboundGatewayTests { +class InboundGatewayTests { private static String topic1 = "testTopic1"; @@ -92,32 +90,37 @@ public class InboundGatewayTests { private static String topic7 = "testTopic7"; - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = - new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6, topic7); + private static EmbeddedKafkaBroker embeddedKafka; - @Rule - public Log4j2LevelAdjuster adjuster = Log4j2LevelAdjuster.trace().categories("org.apache.kafka.clients", - "org.springframework.kafka", "org.springframework.integration"); + @BeforeAll + static void setup() { + embeddedKafka = new EmbeddedKafkaBroker(1, true, + topic1, topic2, topic3, topic4, topic5, topic6, topic7); + embeddedKafka.afterPropertiesSet(); + } + + @AfterAll + static void tearDown() { + embeddedKafka.destroy(); + } @Test - public void testInbound() throws Exception { - EmbeddedKafkaBroker embedded = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); + void testInbound() throws Exception { Map consumerProps = - KafkaTestUtils.consumerProps("replyHandler1", "false", embedded); + KafkaTestUtils.consumerProps("replyHandler1", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); Consumer consumer = cf2.createConsumer(); - embedded.consumeFromAnEmbeddedTopic(consumer, topic2); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic2); - Map props = KafkaTestUtils.consumerProps("test1", "false", embedded); + Map props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); containerProps.setIdleEventInterval(100L); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); - Map senderProps = KafkaTestUtils.producerProps(embedded); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic1); @@ -177,21 +180,20 @@ public class InboundGatewayTests { } @Test - public void testInboundErrorRecover() { - EmbeddedKafkaBroker broker = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); - Map consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", broker); + void testInboundErrorRecover() { + Map consumerProps = KafkaTestUtils.consumerProps("replyHandler2", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); Consumer consumer = cf2.createConsumer(); - broker.consumeFromAnEmbeddedTopic(consumer, topic4); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic4); - Map props = KafkaTestUtils.consumerProps("test2", "false", broker); + Map props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic3); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); - Map senderProps = KafkaTestUtils.producerProps(broker); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic3); @@ -257,21 +259,20 @@ public class InboundGatewayTests { } @Test - public void testInboundRetryErrorRecover() { - EmbeddedKafkaBroker embedded = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); - Map consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embedded); + void testInboundRetryErrorRecover() { + Map consumerProps = KafkaTestUtils.consumerProps("replyHandler3", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); Consumer consumer = cf2.createConsumer(); - embedded.consumeFromAnEmbeddedTopic(consumer, topic6); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, topic6); - Map props = KafkaTestUtils.consumerProps("test3", "false", embedded); + Map props = KafkaTestUtils.consumerProps("test3", "false", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic5); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); - Map senderProps = KafkaTestUtils.producerProps(embedded); + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(topic5); @@ -342,8 +343,7 @@ public class InboundGatewayTests { } @Test - public void testInboundRetryErrorRecoverWithoutRecocveryCallback() throws Exception { - EmbeddedKafkaBroker embeddedKafka = InboundGatewayTests.embeddedKafka.getEmbeddedKafka(); + void testInboundRetryErrorRecoverWithoutRecocveryCallback() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("replyHandler4", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf2 = new DefaultKafkaConsumerFactory<>(consumerProps); 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 1181723efa..1fc76585d4 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 @@ -46,8 +46,9 @@ import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.StaticMessageHeaderAccessor; @@ -77,7 +78,6 @@ import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.kafka.support.converter.StringJsonMessageConverter; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.ContainerTestUtils; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; @@ -102,7 +102,7 @@ import org.springframework.retry.support.RetryTemplate; * @since 2.0 * */ -public class MessageDrivenAdapterTests { +class MessageDrivenAdapterTests { private static String topic1 = "testTopic1"; @@ -116,14 +116,22 @@ public class MessageDrivenAdapterTests { private static String topic6 = "testTopic6"; - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = - new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); + private static EmbeddedKafkaBroker embeddedKafka; - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + @BeforeAll + static void setup() { + embeddedKafka = new EmbeddedKafkaBroker(1, true, + topic1, topic2, topic3, topic4, topic5, topic6); + embeddedKafka.afterPropertiesSet(); + } + + @AfterAll + static void tearDown() { + embeddedKafka.destroy(); + } @Test - public void testInboundRecord() { + void testInboundRecord() { Map props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -208,7 +216,7 @@ public class MessageDrivenAdapterTests { } @Test - public void testInboundRecordRetryRecover() { + void testInboundRecordRetryRecover() { Map props = KafkaTestUtils.consumerProps("test4", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -266,7 +274,7 @@ public class MessageDrivenAdapterTests { * to the consumer. */ @Test - public void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception { + void testInboundRecordRetryRecoverWithoutRecoveryCallback() throws Exception { Map props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -316,7 +324,7 @@ public class MessageDrivenAdapterTests { } @Test - public void testInboundRecordNoRetryRecover() { + void testInboundRecordNoRetryRecover() { Map props = KafkaTestUtils.consumerProps("test5", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -368,7 +376,7 @@ public class MessageDrivenAdapterTests { } @Test - public void testInboundBatch() throws Exception { + void testInboundBatch() throws Exception { Map props = KafkaTestUtils.consumerProps("test2", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -391,6 +399,7 @@ public class MessageDrivenAdapterTests { @Override public Message toMessage(List> records, Acknowledgment acknowledgment, Consumer consumer, Type type) { + Message message = super.toMessage(records, acknowledgment, consumer, type); return MessageBuilder.fromMessage(message).setHeader("testHeader", "testValue").build(); } @@ -431,6 +440,7 @@ public class MessageDrivenAdapterTests { @Override public Message toMessage(List> records, Acknowledgment acknowledgment, Consumer consumer, Type payloadType) { + throw new RuntimeException("testError"); } @@ -453,7 +463,7 @@ public class MessageDrivenAdapterTests { } @Test - public void testInboundJson() { + void testInboundJson() { Map props = KafkaTestUtils.consumerProps("test3", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -495,7 +505,7 @@ public class MessageDrivenAdapterTests { } @Test - public void testInboundJsonWithPayload() { + void testInboundJsonWithPayload() { Map props = KafkaTestUtils.consumerProps("test6", "true", embeddedKafka); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); @@ -503,7 +513,8 @@ public class MessageDrivenAdapterTests { KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); - KafkaMessageDrivenChannelAdapter adapter = Kafka.messageDrivenChannelAdapter(container, ListenerMode.record) + KafkaMessageDrivenChannelAdapter adapter = Kafka + .messageDrivenChannelAdapter(container, ListenerMode.record) .recordMessageConverter(new StringJsonMessageConverter()) .payloadType(Foo.class) .get(); @@ -542,7 +553,7 @@ public class MessageDrivenAdapterTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - public void testPauseResume() throws Exception { + void testPauseResume() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); Consumer consumer = mock(Consumer.class); given(cf.createConsumer(isNull(), eq("clientId"), isNull(), any())).willReturn(consumer); @@ -597,14 +608,14 @@ public class MessageDrivenAdapterTests { adapter.stop(); } - public static class Foo { + static class Foo { private String bar; - public Foo() { + Foo() { } - public Foo(String bar) { + Foo(String bar) { this.bar = bar; } @@ -637,14 +648,11 @@ public class MessageDrivenAdapterTests { } Foo other = (Foo) obj; if (this.bar == null) { - if (other.bar != null) { - return false; - } + return other.bar == null; } - else if (!this.bar.equals(other.bar)) { - return false; + else { + return this.bar.equals(other.bar); } - return true; } } 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 49cb4f9fc2..5d27ec5a2d 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 @@ -26,15 +26,15 @@ import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.common.TopicPartition; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.ConsumerProperties; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; @@ -46,17 +46,25 @@ import org.springframework.messaging.Message; * @since 3.0.1 * */ -public class MessageSourceIntegrationTests { +class MessageSourceIntegrationTests { - public static final String TOPIC1 = "MessageSourceIntegrationTests1"; + private static final String TOPIC1 = "MessageSourceIntegrationTests1"; - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, 1, TOPIC1); + private static EmbeddedKafkaBroker embeddedKafka; - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + @BeforeAll + static void setup() { + embeddedKafka = new EmbeddedKafkaBroker(1, true, 1, TOPIC1); + embeddedKafka.afterPropertiesSet(); + } + + @AfterAll + static void tearDown() { + embeddedKafka.destroy(); + } @Test - public void testSource() throws Exception { + void testSource() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("testSource", "false", embeddedKafka); consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); @@ -64,7 +72,7 @@ public class MessageSourceIntegrationTests { DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProps); ConsumerProperties consumerProperties = new ConsumerProperties(TOPIC1); consumerProperties.getKafkaConsumerProperties() - .setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "2"); + .setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "2"); final CountDownLatch assigned = new CountDownLatch(1); consumerProperties.setConsumerRebalanceListener(new ConsumerRebalanceListener() { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java index 713fbbaf87..2f90beb287 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageSourceTests.java @@ -17,8 +17,8 @@ package org.springframework.integration.kafka.inbound; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatThrownBy; -import static org.assertj.core.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyCollection; import static org.mockito.ArgumentMatchers.anyLong; @@ -64,7 +64,7 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.clients.consumer.OffsetResetStrategy; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.record.TimestampType; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.InOrder; @@ -87,14 +87,16 @@ import org.springframework.messaging.Message; /** * @author Gary Russell * @author Anshul Mehra + * @author Artem Bilan + * * @since 3.0.1 * */ -public class MessageSourceTests { +class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testIllegalArgs() { + void testIllegalArgs() { ConsumerFactory consumerFactory = mock(ConsumerFactory.class); assertThatThrownBy(() -> new KafkaMessageSource(consumerFactory, new ConsumerProperties((Pattern) null))) .isInstanceOf(IllegalArgumentException.class) @@ -103,7 +105,7 @@ public class MessageSourceTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - public void testConsumerAwareRebalanceListener() { + void testConsumerAwareRebalanceListener() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); List assigned = Collections.singletonList(topicPartition); @@ -123,6 +125,7 @@ public class MessageSourceTests { AtomicBoolean partitionsRevokedCalled = new AtomicBoolean(); AtomicReference partitionsRevokedConsumer = new AtomicReference<>(); consumerProperties.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() { + @Override public void onPartitionsRevokedAfterCommit(Consumer cons, Collection partitions) { partitionsRevokedCalled.getAndSet(true); @@ -152,7 +155,7 @@ public class MessageSourceTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - public void testRebalanceListener() { + void testRebalanceListener() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); List assigned = Collections.singletonList(topicPartition); @@ -195,17 +198,17 @@ public class MessageSourceTests { } @Test - public void testAckSyncCommits() { + void testAckSyncCommits() { testAckCommon(true, false); } @Test - public void testAckSyncCommitsTimeout() { + void testAckSyncCommitsTimeout() { testAckCommon(true, false); } @Test - public void testAckAsyncCommits() { + void testAckAsyncCommits() { testAckCommon(false, false); } @@ -269,7 +272,7 @@ public class MessageSourceTests { assertThat(received).isNotNull(); assertThat(received.getHeaders().get(KafkaHeaders.RAW_DATA)).isInstanceOf(ConsumerRecord.class); assertThat(received.getHeaders().get(IntegrationMessageHeaderAccessor.SOURCE_DATA)) - .isSameAs(received.getHeaders().get(KafkaHeaders.RAW_DATA)); + .isSameAs(received.getHeaders().get(KafkaHeaders.RAW_DATA)); StaticMessageHeaderAccessor.getAcknowledgmentCallback(received) .acknowledge(AcknowledgmentCallback.Status.ACCEPT); received = source.receive(); @@ -337,7 +340,7 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testAckOutOfOrder() { + void testAckOutOfOrder() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); willAnswer(i -> { @@ -427,7 +430,7 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testNack() { + void testNack() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); willAnswer(i -> { @@ -442,7 +445,7 @@ public class MessageSourceTests { }).given(consumer).pause(anyCollection()); willAnswer(i -> paused.get()).given(consumer).paused(); Map> records1 = new LinkedHashMap<>(); - records1.put(topicPartition, Arrays.asList( + records1.put(topicPartition, Collections.singletonList( new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); ConsumerRecords cr1 = new ConsumerRecords(records1); Map> records2 = new LinkedHashMap<>(); @@ -496,7 +499,7 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testNackWithLaterInflight() { + void testNackWithLaterInflight() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); willAnswer(i -> { @@ -511,7 +514,7 @@ public class MessageSourceTests { }).given(consumer).pause(anyCollection()); willAnswer(i -> paused.get()).given(consumer).paused(); Map> records1 = new LinkedHashMap<>(); - records1.put(topicPartition, Arrays.asList( + records1.put(topicPartition, Collections.singletonList( new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); ConsumerRecords cr1 = new ConsumerRecords(records1); Map> records2 = new LinkedHashMap<>(); @@ -582,7 +585,7 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testMaxPollRecords() { + void testMaxPollRecords() { KafkaMessageSource source = new KafkaMessageSource(new DefaultKafkaConsumerFactory<>(Collections.emptyMap()), new ConsumerProperties("topic")); assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class) @@ -591,20 +594,17 @@ public class MessageSourceTests { Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2)), new ConsumerProperties("topic")); assertThat((TestUtils.getPropertyValue(source, "consumerFactory.configs", Map.class) .get(ConsumerConfig.MAX_POLL_RECORDS_CONFIG))).isEqualTo(1); - try { - new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) { - }), new ConsumerProperties("topic")); - fail("Expected exception"); - } - catch (IllegalArgumentException e) { - assertThat(e.getMessage()).contains(ConsumerConfig.MAX_POLL_RECORDS_CONFIG); - } + assertThatIllegalArgumentException() + .isThrownBy(() -> + new KafkaMessageSource((new DefaultKafkaConsumerFactory(Collections.emptyMap()) { }), + new ConsumerProperties("topic"))) + .withMessageContaining(ConsumerConfig.MAX_POLL_RECORDS_CONFIG); } @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testPollTimeouts() { + void testPollTimeouts() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); List assigned = Collections.singletonList(topicPartition); @@ -615,12 +615,12 @@ public class MessageSourceTests { }).given(consumer).subscribe(anyCollection(), any(ConsumerRebalanceListener.class)); Map> records1 = new LinkedHashMap<>(); - records1.put(topicPartition, Arrays.asList( + records1.put(topicPartition, Collections.singletonList( new ConsumerRecord("foo", 0, 0L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); ConsumerRecords cr1 = new ConsumerRecords(records1); given(consumer.poll(Duration.of(20 * 5000, ChronoUnit.MILLIS))).willReturn(cr1, ConsumerRecords.EMPTY); Map> records2 = new LinkedHashMap<>(); - records2.put(topicPartition, Arrays.asList( + records2.put(topicPartition, Collections.singletonList( new ConsumerRecord("foo", 0, 1L, 0L, TimestampType.NO_TIMESTAMP_TYPE, 0, 0, 0, null, "foo"))); ConsumerRecords cr2 = new ConsumerRecords(records2); given(consumer.poll(Duration.of(5000, ChronoUnit.MILLIS))).willReturn(cr2, ConsumerRecords.EMPTY); @@ -665,7 +665,7 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testAllowMulti() { + void testAllowMulti() { Consumer consumer = mock(Consumer.class); TopicPartition topicPartition = new TopicPartition("foo", 0); List assigned = Collections.singletonList(topicPartition); @@ -730,7 +730,7 @@ public class MessageSourceTests { @SuppressWarnings("unchecked") @Test - public void testTopicPatternBasedMessageSource() { + void testTopicPatternBasedMessageSource() { MockConsumer consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST); TopicPartition topicPartition1 = new TopicPartition("abc_foo", 0); TopicPartition topicPartition2 = new TopicPartition("abc_foo", 1); @@ -747,7 +747,9 @@ public class MessageSourceTests { willReturn(Collections.singletonMap(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1)).given(consumerFactory) .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer); - KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, new ConsumerProperties(Pattern.compile("[a-zA-Z0-9_]*?foo"))); + KafkaMessageSource source = new KafkaMessageSource<>(consumerFactory, + new ConsumerProperties(Pattern + .compile("[a-zA-Z0-9_]*?foo"))); source.setRawMessageHeader(true); source.start(); // force consumer creation @@ -789,7 +791,7 @@ public class MessageSourceTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testStaticPartitionAssignment() { + void testStaticPartitionAssignment() { MockConsumer consumer = spy(new MockConsumer<>(OffsetResetStrategy.EARLIEST)); TopicPartition beginning = new TopicPartition("foo", 0); @@ -818,9 +820,11 @@ public class MessageSourceTests { .getConfigurationProperties(); given(consumerFactory.createConsumer(isNull(), anyString(), isNull(), any())).willReturn(consumer); - TopicPartitionOffset beginningTpo = new TopicPartitionOffset(beginning, null, TopicPartitionOffset.SeekPosition.BEGINNING); + TopicPartitionOffset beginningTpo = new TopicPartitionOffset(beginning, null, + TopicPartitionOffset.SeekPosition.BEGINNING); TopicPartitionOffset endTpo = new TopicPartitionOffset(end, null, TopicPartitionOffset.SeekPosition.END); - TopicPartitionOffset timestampTpo = new TopicPartitionOffset(timestamp, null, TopicPartitionOffset.SeekPosition.TIMESTAMP); + TopicPartitionOffset timestampTpo = new TopicPartitionOffset(timestamp, null, + TopicPartitionOffset.SeekPosition.TIMESTAMP); TopicPartitionOffset negativeOffsetTpo = new TopicPartitionOffset(negativeOffset, -1L, null); TopicPartitionOffset negativeRelativeToCurrentTpo = new TopicPartitionOffset(negativeRelativeToCurrent.topic(), negativeRelativeToCurrent.partition(), -1L, true); @@ -865,11 +869,11 @@ public class MessageSourceTests { Message message; Set expected = Stream.of( p0r0, p0r1, p0r2, p0r3, // Seek to beginning - p1r3, // Seek to end - p2r1, p2r2, p2r3, // Null offset and SeekPosition.TIMESTAMP results in no change in position - p3r3, // Negative offset ends up in seek to end - p4r2, p4r3, // Negative offset with relative to current(3) - p5r2, p5r3 // Positive offset with relative to current(1) + p1r3, // Seek to end + p2r1, p2r2, p2r3, // Null offset and SeekPosition.TIMESTAMP results in no change in position + p3r3, // Negative offset ends up in seek to end + p4r2, p4r3, // Negative offset with relative to current(3) + p5r2, p5r3 // Positive offset with relative to current(1) ).map(ConsumerRecord::value).collect(Collectors.toSet()); Set received = new HashSet<>(); while ((message = source.receive()) != null) { @@ -902,4 +906,5 @@ public class MessageSourceTests { inOrder.verify(consumer).close(anyLong(), any(TimeUnit.class)); inOrder.verifyNoMoreInteractions(); } + } 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 c576919c5f..8a7ba0c883 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 @@ -17,7 +17,7 @@ package org.springframework.integration.kafka.outbound; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.BDDMockito.given; @@ -56,10 +56,9 @@ import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.TopicPartition; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.InOrder; @@ -87,7 +86,6 @@ import org.springframework.kafka.support.KafkaNull; import org.springframework.kafka.support.SendResult; import org.springframework.kafka.support.TransactionSupport; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.kafka.transaction.KafkaTransactionManager; import org.springframework.messaging.Message; @@ -108,7 +106,7 @@ import org.springframework.util.concurrent.SettableListenableFuture; * * @since 2.0 */ -public class KafkaProducerMessageHandlerTests { +class KafkaProducerMessageHandlerTests { private static String topic1 = "testTopic1out"; @@ -122,16 +120,15 @@ public class KafkaProducerMessageHandlerTests { private static String topic6 = "testTopic6in"; - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = - new EmbeddedKafkaRule(1, true, topic1, topic2, topic3, topic4, topic5, topic6); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; - @BeforeClass - public static void setUp() { + @BeforeAll + static void setup() { + embeddedKafka = new EmbeddedKafkaBroker(1, true, + topic1, topic2, topic3, topic4, topic5, topic6); + embeddedKafka.afterPropertiesSet(); Map consumerProps = KafkaTestUtils.consumerProps("testOut", "true", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); ConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); @@ -139,13 +136,14 @@ public class KafkaProducerMessageHandlerTests { embeddedKafka.consumeFromAllEmbeddedTopics(consumer); } - @AfterClass - public static void tearDown() { + @AfterAll + static void tearDown() { consumer.close(); + embeddedKafka.destroy(); } @Test - public void testOutbound() throws Exception { + void testOutbound() { DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); @@ -171,7 +169,7 @@ public class KafkaProducerMessageHandlerTests { .build(); handler.handleMessage(message); record = KafkaTestUtils.getSingleRecord(consumer, topic1); - assertThat(record).has(key((Integer) null)); + assertThat(record).has(key(null)); assertThat(record).has(partition(0)); assertThat(record).has(value("bar")); @@ -180,7 +178,7 @@ public class KafkaProducerMessageHandlerTests { .build(); handler.handleMessage(message); record = KafkaTestUtils.getSingleRecord(consumer, topic1); - assertThat(record).has(key((Integer) null)); + assertThat(record).has(key(null)); assertThat(record).has(value("baz")); handler.setPartitionIdExpression(new SpelExpressionParser().parseExpression("headers['kafka_partitionId']")); @@ -201,7 +199,7 @@ public class KafkaProducerMessageHandlerTests { } @Test - public void testOutboundWithTimestamp() { + void testOutboundWithTimestamp() { DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); @@ -232,7 +230,7 @@ public class KafkaProducerMessageHandlerTests { } @Test - public void testOutboundWithTimestampExpression() { + void testOutboundWithTimestampExpression() { DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); @@ -271,7 +269,7 @@ public class KafkaProducerMessageHandlerTests { } @Test - public void testOutboundWithAsyncResults() { + void testOutboundWithAsyncResults() { DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); @@ -331,7 +329,7 @@ public class KafkaProducerMessageHandlerTests { } @Test - public void testOutboundWithCustomHeaderMapper() throws Exception { + void testOutboundWithCustomHeaderMapper() { DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); @@ -353,7 +351,7 @@ public class KafkaProducerMessageHandlerTests { } @Test - public void testOutboundGateway() throws Exception { + void testOutboundGateway() throws Exception { ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>( KafkaTestUtils.consumerProps(topic5, "false", embeddedKafka)); ContainerProperties containerProperties = new ContainerProperties(topic6); @@ -376,7 +374,8 @@ public class KafkaProducerMessageHandlerTests { DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); - ReplyingKafkaTemplate template = new ReplyingKafkaTemplate<>(producerFactory, container); + ReplyingKafkaTemplate template = + new ReplyingKafkaTemplate<>(producerFactory, container); template.start(); assertThat(assigned.await(30, TimeUnit.SECONDS)).isTrue(); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); @@ -407,37 +406,30 @@ public class KafkaProducerMessageHandlerTests { assertThat(reply.getHeaders().get(KafkaHeaders.TOPIC)).isNull(); assertThat(reply.getHeaders().get(KafkaHeaders.CORRELATION_ID)).isNull(); - message = MessageBuilder.withPayload("foo") + final Message messageToHandle1 = MessageBuilder.withPayload("foo") .setHeader(KafkaHeaders.TOPIC, topic5) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) .setHeader(KafkaHeaders.PARTITION_ID, 1) .setHeader(KafkaHeaders.REPLY_TOPIC, "bad") .build(); - try { - handler.handleMessage(message); - fail("Expected exception"); - } - catch (MessageHandlingException e) { - assertThat(e.getCause().getMessage()) - .isEqualTo("The reply topic header [bad] does not match any reply container topic: " - + "[" + topic6 + "]"); - } - message = MessageBuilder.withPayload("foo") + + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> handler.handleMessage(messageToHandle1)) + .withMessageContaining("The reply topic header [bad] does not match any reply container topic: " + + "[" + topic6 + "]"); + + final Message messageToHandle2 = MessageBuilder.withPayload("foo") .setHeader(KafkaHeaders.TOPIC, topic5) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) .setHeader(KafkaHeaders.PARTITION_ID, 1) .setHeader(KafkaHeaders.REPLY_PARTITION, 999) .build(); - try { - handler.handleMessage(message); - fail("Expected exception"); - } - catch (MessageHandlingException e) { - assertThat(e.getCause().getMessage()) - .isEqualTo( - "The reply partition header [999] does not match any reply container partition for topic [" - + topic6 + "]: [0, 1]"); - } + + assertThatExceptionOfType(MessageHandlingException.class) + .isThrownBy(() -> handler.handleMessage(messageToHandle2)) + .withMessageContaining("The reply partition header [999] " + + "does not match any reply container partition for topic [" + + topic6 + "]: [0, 1]"); template.stop(); // discard from the test consumer @@ -448,7 +440,7 @@ public class KafkaProducerMessageHandlerTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testTransaction() { + void testTransaction() { ProducerFactory pf = mock(ProducerFactory.class); given(pf.transactionCapable()).willReturn(true); Producer producer = mock(Producer.class); @@ -472,7 +464,7 @@ public class KafkaProducerMessageHandlerTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - public void testConsumeAndProduceTransaction() throws Exception { + void testConsumeAndProduceTransaction() throws Exception { Consumer mockConsumer = mock(Consumer.class); final TopicPartition topicPartition = new TopicPartition("foo", 0); willAnswer(i -> { @@ -507,8 +499,7 @@ public class KafkaProducerMessageHandlerTests { transactionalIds.add(TransactionSupport.getTransactionIdSuffix()); return producer; }).given(pf).createProducer(isNull()); - KafkaTransactionManager tm = new KafkaTransactionManager(pf); - PlatformTransactionManager ptm = tm; + PlatformTransactionManager ptm = new KafkaTransactionManager(pf); ContainerProperties props = new ContainerProperties("foo"); props.setGroupId("group"); props.setTransactionManager(ptm); @@ -549,7 +540,7 @@ public class KafkaProducerMessageHandlerTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Test - public void testTransactionTxIdOverride() { + void testTransactionTxIdOverride() { Producer producer = mock(Producer.class); AtomicReference txId = new AtomicReference<>(); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory(Collections.emptyMap()) { @@ -584,7 +575,7 @@ public class KafkaProducerMessageHandlerTests { @SuppressWarnings({ "unchecked", "rawtypes" }) @Test - public void testConsumeAndProduceTransactionTxIdOverride() throws Exception { + void testConsumeAndProduceTransactionTxIdOverride() throws Exception { Consumer mockConsumer = mock(Consumer.class); final TopicPartition topicPartition = new TopicPartition("foo", 0); willAnswer(i -> { 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 63aedd9ad4..9d50d232d1 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 @@ -27,9 +27,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig import org.apache.kafka.clients.consumer.ConsumerRebalanceListener import org.apache.kafka.clients.producer.ProducerConfig import org.apache.kafka.common.TopicPartition -import org.junit.ClassRule import org.junit.Test -import org.junit.runner.RunWith import org.springframework.beans.factory.annotation.Autowired import org.springframework.beans.factory.annotation.Qualifier import org.springframework.context.annotation.Bean @@ -62,7 +60,8 @@ import org.springframework.kafka.requestreply.ReplyingKafkaTemplate import org.springframework.kafka.support.Acknowledgment import org.springframework.kafka.support.DefaultKafkaHeaderMapper import org.springframework.kafka.support.KafkaHeaders -import org.springframework.kafka.test.rule.EmbeddedKafkaRule +import org.springframework.kafka.test.EmbeddedKafkaBroker +import org.springframework.kafka.test.context.EmbeddedKafka import org.springframework.kafka.test.utils.KafkaTestUtils import org.springframework.messaging.Message import org.springframework.messaging.MessageChannel @@ -72,7 +71,7 @@ import org.springframework.messaging.support.ErrorMessage import org.springframework.messaging.support.GenericMessage import org.springframework.retry.support.RetryTemplate import org.springframework.test.annotation.DirtiesContext -import org.springframework.test.context.junit4.SpringRunner +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig import java.time.Duration import java.util.concurrent.CountDownLatch import java.util.concurrent.TimeUnit @@ -85,296 +84,298 @@ import java.util.stream.Stream * @since 3.0.3 */ -@RunWith(SpringRunner::class) +@SpringJUnitConfig @DirtiesContext +@EmbeddedKafka(topics = [KafkaDslKotlinTests.TEST_TOPIC1, KafkaDslKotlinTests.TEST_TOPIC2, + KafkaDslKotlinTests.TEST_TOPIC3, KafkaDslKotlinTests.TEST_TOPIC4, KafkaDslKotlinTests.TEST_TOPIC5]) class KafkaDslKotlinTests { - companion object { + companion object { - const val TEST_TOPIC1 = "test-topic1" + const val TEST_TOPIC1 = "test-topic1" - const val TEST_TOPIC2 = "test-topic2" + const val TEST_TOPIC2 = "test-topic2" - const val TEST_TOPIC3 = "test-topic3" + const val TEST_TOPIC3 = "test-topic3" - const val TEST_TOPIC4 = "test-topic4" + const val TEST_TOPIC4 = "test-topic4" - const val TEST_TOPIC5 = "test-topic5" + const val TEST_TOPIC5 = "test-topic5" + + } - @ClassRule - @JvmField - var embeddedKafka = EmbeddedKafkaRule(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5) - } + @Autowired + @Qualifier("sendToKafkaFlow.input") + private lateinit var sendToKafkaFlowInput: MessageChannel + + @Autowired + private lateinit var listeningFromKafkaResults1: PollableChannel + + @Autowired + private lateinit var listeningFromKafkaResults2: PollableChannel + + @Autowired + @Qualifier("kafkaProducer1.handler") + private lateinit var kafkaProducer1: KafkaProducerMessageHandler<*, *> + + @Autowired + @Qualifier("kafkaProducer2.handler") + private lateinit var kafkaProducer2: KafkaProducerMessageHandler<*, *> + + @Autowired + private lateinit var errorChannel: PollableChannel + + @Autowired(required = false) + @Qualifier("topic1ListenerContainer") + private lateinit var messageListenerContainer: MessageListenerContainer + + @Autowired(required = false) + @Qualifier("kafkaTemplate:test-topic1") + private lateinit var kafkaTemplateTopic1: KafkaTemplate + + @Autowired(required = false) + @Qualifier("kafkaTemplate:test-topic2") + private lateinit var kafkaTemplateTopic2: KafkaTemplate<*, *> + + @Autowired + private lateinit var mapper: DefaultKafkaHeaderMapper + + @Autowired + private lateinit var config: ContextConfiguration + + @Autowired + private lateinit var gate: Gate + + @Test + fun testKafkaAdapters() { + this.sendToKafkaFlowInput.send(GenericMessage("foo", hashMapOf("foo" to "bar"))) + + assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper) + + for (i in 0..99) { + val receive = this.listeningFromKafkaResults1.receive(20000) + assertThat(receive).isNotNull() + assertThat(receive!!.payload).isEqualTo("FOO") + val headers = receive.headers + assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue() + val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java) + acknowledgment?.acknowledge() + assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC1) + assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1) + assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0) + assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong()) + assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME") + assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048633L) + assertThat(headers["foo"]).isEqualTo("bar") + } + + for (i in 0..99) { + val receive = this.listeningFromKafkaResults2.receive(20000) + assertThat(receive).isNotNull() + assertThat(receive!!.payload).isEqualTo("FOO") + val headers = receive.headers + assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue() + val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java) + acknowledgment?.acknowledge() + assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC2) + assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1) + assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0) + assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong()) + assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME") + assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048644L) + } + + val message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build() + + this.sendToKafkaFlowInput.send(message) + + assertThat(this.listeningFromKafkaResults1.receive(10)).isNull() + + val error = this.errorChannel.receive(10000) + + assertThat(error).isNotNull().isInstanceOf(ErrorMessage::class.java) + + val payload = error?.payload + + assertThat(payload).isNotNull().isInstanceOf(MessageRejectedException::class.java) + + assertThat(this.messageListenerContainer).isNotNull() + assertThat(this.kafkaTemplateTopic1).isNotNull() + assertThat(this.kafkaTemplateTopic2).isNotNull() + + this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo") + assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue() + assertThat(this.config.fromSource).isEqualTo("foo") + } + + @Test + fun testGateways() { + assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)) + assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO") + } + + @Configuration + @EnableIntegration + @EnableKafka + class ContextConfiguration { + + val sourceFlowLatch = CountDownLatch(1) + + val replyContainerLatch = CountDownLatch(1) + + var fromSource: Any? = null + + @Autowired + private lateinit var embeddedKafka: EmbeddedKafkaBroker + + @Bean + fun consumerFactory(): ConsumerFactory { + val props = KafkaTestUtils.consumerProps("test1", "false", this.embeddedKafka) + props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") + return DefaultKafkaConsumerFactory(props) + } + + @Bean + fun errorChannel() = QueueChannel() + + @Bean + fun topic1ListenerFromKafkaFlow() = + IntegrationFlows.from( + Kafka.messageDrivenChannelAdapter(consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) + .configureListenerContainer { + it.ackMode(ContainerProperties.AckMode.MANUAL) + .id("topic1ListenerContainer") + } + .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(), + RawRecordHeaderErrorMessageStrategy())) + .retryTemplate(RetryTemplate()) + .filterInRetry(true)) + .filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 }, + { f -> f.throwExceptionOnRejection(true) }) + .transform { it.toUpperCase() } + .channel { c -> c.queue("listeningFromKafkaResults1") } + .get() + + @Bean + fun topic2ListenerFromKafkaFlow() = + IntegrationFlows.from( + Kafka.messageDrivenChannelAdapter(consumerFactory(), + KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2) + .configureListenerContainer { it.ackMode(ContainerProperties.AckMode.MANUAL) } + .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(), + RawRecordHeaderErrorMessageStrategy())) + .retryTemplate(RetryTemplate()) + .filterInRetry(true)) + .filter(Message::class.java, + { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 }, + { it.throwExceptionOnRejection(true) }) + .transform { it.toUpperCase() } + .channel { c -> c.queue("listeningFromKafkaResults2") } + .get() + + @Bean + fun producerFactory(): DefaultKafkaProducerFactory { + val props = KafkaTestUtils.producerProps(this.embeddedKafka) + props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000") + return DefaultKafkaProducerFactory(props) + } + + @Bean + fun sendToKafkaFlow() = + IntegrationFlow { f -> + f.split({ p -> Stream.generate { p }.limit(101) }, null) + .publishSubscribeChannel { c -> + c.subscribe { sf -> + sf.handle( + kafkaMessageHandler(producerFactory(), TEST_TOPIC1) + .timestampExpression("T(Long).valueOf('1487694048633')") + ) { it.id("kafkaProducer1") } + } + .subscribe { sf -> + sf.handle( + kafkaMessageHandler(producerFactory(), TEST_TOPIC2) + .timestamp { 1487694048644L } + ) { it.id("kafkaProducer2") } + } + } + } + + @Bean + fun mapper() = DefaultKafkaHeaderMapper() + + private fun kafkaMessageHandler(producerFactory: ProducerFactory, topic: String) = + Kafka.outboundChannelAdapter(producerFactory) + .messageKey { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } + .headerMapper(mapper()) + .sync(true) + .partitionId { _ -> 0 } + .topicExpression("headers[kafka_topic] ?: '$topic'") + .configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") } - @Autowired - @Qualifier("sendToKafkaFlow.input") - private lateinit var sendToKafkaFlowInput: MessageChannel + @Bean + fun sourceFlow() = + IntegrationFlows + .from(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3))) + { e -> e.poller(Pollers.fixedDelay(100)) } + .handle { p -> + this.fromSource = p.getPayload() + this.sourceFlowLatch.countDown() + } + .get() - @Autowired - private lateinit var listeningFromKafkaResults1: PollableChannel + @Bean + fun replyingKafkaTemplate() = + ReplyingKafkaTemplate(producerFactory(), replyContainer()) + .also { + it.setDefaultReplyTimeout(Duration.ofSeconds(30)) + } - @Autowired - private lateinit var listeningFromKafkaResults2: PollableChannel + @Bean + fun outboundGateFlow() = + IntegrationFlows.from(Gate::class.java) + .handle(Kafka.outboundGateway(replyingKafkaTemplate()) + .sync(true)) + .get() - @Autowired - @Qualifier("kafkaProducer1.handler") - private lateinit var kafkaProducer1: KafkaProducerMessageHandler<*, *> + private fun replyContainer(): GenericMessageListenerContainer { + val containerProperties = ContainerProperties(TEST_TOPIC5) + containerProperties.groupId = "outGate" + containerProperties.consumerRebalanceListener = object : ConsumerRebalanceListener { - @Autowired - @Qualifier("kafkaProducer2.handler") - private lateinit var kafkaProducer2: KafkaProducerMessageHandler<*, *> + override fun onPartitionsRevoked(partitions: Collection) { + // empty + } - @Autowired - private lateinit var errorChannel: PollableChannel + override fun onPartitionsAssigned(partitions: Collection) { + this@ContextConfiguration.replyContainerLatch.countDown() + } - @Autowired(required = false) - @Qualifier("topic1ListenerContainer") - private lateinit var messageListenerContainer: MessageListenerContainer + } + return KafkaMessageListenerContainer(consumerFactory(), containerProperties) + } - @Autowired(required = false) - @Qualifier("kafkaTemplate:test-topic1") - private lateinit var kafkaTemplateTopic1: KafkaTemplate + @Bean + fun serverGateway() = + IntegrationFlows.from( + Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) + .transform { it.toUpperCase() } + .get() - @Autowired(required = false) - @Qualifier("kafkaTemplate:test-topic2") - private lateinit var kafkaTemplateTopic2: KafkaTemplate<*, *> + private fun containerProperties() = + ContainerProperties(TEST_TOPIC4) + .also { + it.groupId = "inGateGroup" + } - @Autowired - private lateinit var mapper: DefaultKafkaHeaderMapper + } - @Autowired - private lateinit var config: ContextConfiguration + interface Gate { - @Autowired - private lateinit var gate: Gate + fun exchange(@Header(KafkaHeaders.TOPIC) topic: String, out: String): String - @Test - fun testKafkaAdapters() { - - this.sendToKafkaFlowInput.send(GenericMessage("foo", hashMapOf("foo" to "bar"))) - - assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper) - - for (i in 0..99) { - val receive = this.listeningFromKafkaResults1.receive(20000) - assertThat(receive).isNotNull() - assertThat(receive!!.payload).isEqualTo("FOO") - val headers = receive.headers - assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue() - val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java) - acknowledgment?.acknowledge() - assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC1) - assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1) - assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0) - assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong()) - assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME") - assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048633L) - assertThat(headers["foo"]).isEqualTo("bar") - } - - for (i in 0..99) { - val receive = this.listeningFromKafkaResults2.receive(20000) - assertThat(receive).isNotNull() - assertThat(receive!!.payload).isEqualTo("FOO") - val headers = receive.headers - assertThat(headers.containsKey(KafkaHeaders.ACKNOWLEDGMENT)).isTrue() - val acknowledgment = headers.get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment::class.java) - acknowledgment?.acknowledge() - assertThat(headers[KafkaHeaders.RECEIVED_TOPIC]).isEqualTo(TEST_TOPIC2) - assertThat(headers[KafkaHeaders.RECEIVED_MESSAGE_KEY]).isEqualTo(i + 1) - assertThat(headers[KafkaHeaders.RECEIVED_PARTITION_ID]).isEqualTo(0) - assertThat(headers[KafkaHeaders.OFFSET]).isEqualTo(i.toLong()) - assertThat(headers[KafkaHeaders.TIMESTAMP_TYPE]).isEqualTo("CREATE_TIME") - assertThat(headers[KafkaHeaders.RECEIVED_TIMESTAMP]).isEqualTo(1487694048644L) - } - - val message = MessageBuilder.withPayload("BAR").setHeader(KafkaHeaders.TOPIC, TEST_TOPIC2).build() - - this.sendToKafkaFlowInput.send(message) - - assertThat(this.listeningFromKafkaResults1.receive(10)).isNull() - - val error = this.errorChannel.receive(10000) - - assertThat(error).isNotNull().isInstanceOf(ErrorMessage::class.java) - - val payload = error?.payload - - assertThat(payload).isNotNull().isInstanceOf(MessageRejectedException::class.java) - - assertThat(this.messageListenerContainer).isNotNull() - assertThat(this.kafkaTemplateTopic1).isNotNull() - assertThat(this.kafkaTemplateTopic2).isNotNull() - - this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo") - assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue() - assertThat(this.config.fromSource).isEqualTo("foo") - } - - @Test - fun testGateways() { - assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)) - assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO") - } - - @Configuration - @EnableIntegration - @EnableKafka - class ContextConfiguration { - - val sourceFlowLatch = CountDownLatch(1) - - val replyContainerLatch = CountDownLatch(1) - - var fromSource: Any? = null - - @Bean - fun consumerFactory(): ConsumerFactory { - val props = KafkaTestUtils.consumerProps("test1", "false", embeddedKafka.embeddedKafka) - props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest") - return DefaultKafkaConsumerFactory(props) - } - - @Bean - fun errorChannel() = QueueChannel() - - @Bean - fun topic1ListenerFromKafkaFlow() = - IntegrationFlows.from( - Kafka.messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) - .configureListenerContainer({ c -> - c.ackMode(ContainerProperties.AckMode.MANUAL) - .id("topic1ListenerContainer") - }) - .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(), - RawRecordHeaderErrorMessageStrategy())) - .retryTemplate(RetryTemplate()) - .filterInRetry(true)) - .filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 }, - { f -> f.throwExceptionOnRejection(true) }) - .transform { it: String -> it.toUpperCase() } - .channel { c -> c.queue("listeningFromKafkaResults1") } - .get() - - @Bean - fun topic2ListenerFromKafkaFlow() = - IntegrationFlows.from( - Kafka.messageDrivenChannelAdapter(consumerFactory(), - KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2) - .configureListenerContainer({ c -> c.ackMode(ContainerProperties.AckMode.MANUAL) }) - .recoveryCallback(ErrorMessageSendingRecoverer(errorChannel(), - RawRecordHeaderErrorMessageStrategy())) - .retryTemplate(RetryTemplate()) - .filterInRetry(true)) - .filter(Message::class.java, { m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer::class.java)!! < 101 }, - { f -> f.throwExceptionOnRejection(true) }) - .transform { it: String -> it.toUpperCase() } - .channel { c -> c.queue("listeningFromKafkaResults2") } - .get() - - @Bean - fun producerFactory(): DefaultKafkaProducerFactory { - val props = KafkaTestUtils.producerProps(embeddedKafka.embeddedKafka) - props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000") - return DefaultKafkaProducerFactory(props) - } - - @Bean - fun sendToKafkaFlow() = - IntegrationFlow { f -> - f.split({ p -> Stream.generate { p }.limit(101) }, null) - .publishSubscribeChannel { c -> - c.subscribe { sf -> - sf.handle( - kafkaMessageHandler(producerFactory(), TEST_TOPIC1) - .timestampExpression("T(Long).valueOf('1487694048633')"), - { e -> e.id("kafkaProducer1") }) - } - .subscribe { sf -> - sf.handle( - kafkaMessageHandler(producerFactory(), TEST_TOPIC2) - .timestamp { _ -> 1487694048644L }, - { e -> e.id("kafkaProducer2") }) - } - } - } - - @Bean - fun mapper() = DefaultKafkaHeaderMapper() - - private fun kafkaMessageHandler(producerFactory: ProducerFactory, topic: String) = - Kafka.outboundChannelAdapter(producerFactory) - .messageKey { m -> m.headers[IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER] } - .headerMapper(mapper()) - .sync(true) - .partitionId { _ -> 0 } - .topicExpression("headers[kafka_topic] ?: '$topic'") - .configureKafkaTemplate { t -> t.id("kafkaTemplate:$topic") } - - - @Bean - fun sourceFlow() = - IntegrationFlows - .from(Kafka.inboundChannelAdapter(consumerFactory(), ConsumerProperties(TEST_TOPIC3))) { e -> e.poller(Pollers.fixedDelay(100)) } - .handle({ p -> - this.fromSource = p.getPayload() - this.sourceFlowLatch.countDown() - }) - .get() - - @Bean - fun replyingKafkaTemplate() = - ReplyingKafkaTemplate(producerFactory(), replyContainer()) - .also { - it.setDefaultReplyTimeout(Duration.ofSeconds(30)) - } - - @Bean - fun outboundGateFlow() = - IntegrationFlows.from(Gate::class.java) - .handle(Kafka.outboundGateway(replyingKafkaTemplate()) - .sync(true)) - .get() - - private fun replyContainer(): GenericMessageListenerContainer { - val containerProperties = ContainerProperties(TEST_TOPIC5) - containerProperties.setGroupId("outGate") - containerProperties.setConsumerRebalanceListener(object : ConsumerRebalanceListener { - - override fun onPartitionsRevoked(partitions: Collection) { - // empty - } - - override fun onPartitionsAssigned(partitions: Collection) { - this@ContextConfiguration.replyContainerLatch.countDown() - } - - }) - return KafkaMessageListenerContainer(consumerFactory(), containerProperties) - } - - @Bean - fun serverGateway() = - IntegrationFlows.from( - Kafka.inboundGateway(consumerFactory(), containerProperties(), producerFactory())) - .transform { it: String -> it.toUpperCase() } - .get() - - private fun containerProperties() = - ContainerProperties(TEST_TOPIC4) - .also { - it.setGroupId("inGateGroup") - } - - } - - interface Gate { - - fun exchange(@Header(KafkaHeaders.TOPIC) topic: String, out: String): String - - } + } }