From 9b8ebe235eddd3a2400146b1118b978e4ab685d7 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 11 Mar 2022 16:50:51 -0500 Subject: [PATCH] JUnit 5 migration for more Kafka binder tests --- .../KafkaBinderMeterRegistryTest.java | 2 +- .../integration/KafkaBinderActuatorTests.java | 42 +++++------------- .../KafkaBinderExtendedPropertiesTest.java | 34 ++++---------- .../KafkaConfigCustomizationTests.java | 37 ++++++---------- .../integration/KafkaNullConverterTest.java | 40 +++++------------ .../KafkaRetryDlqBinderOrContainerTests.java | 4 +- .../ProducerOnlyTransactionTests.java | 44 ++++++------------- ...eKafkaBinderTopicPropertiesUpdateTest.java | 33 +++++--------- ...dKafkaBinderTopicPropertiesUpdateTest.java | 12 ++--- .../KafkaBinderTopicPropertiesUpdateTest.java | 8 ++-- .../ConsumerProducerTransactionTests.java | 41 ++++------------- 11 files changed, 89 insertions(+), 208 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java index 9f438b1d3..88cf55bb6 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/bootstrap/KafkaBinderMeterRegistryTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-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. diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java index ce5e1d105..c16415132 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderActuatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -22,12 +22,9 @@ import java.util.function.Consumer; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.binder.MeterBinder; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Ignore; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; @@ -48,10 +45,10 @@ import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AbstractMessageListenerContainer; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; import static org.assertj.core.api.Assertions.assertThat; @@ -64,36 +61,19 @@ import static org.assertj.core.api.Assertions.assertThat; * * @since 2.0 */ -@RunWith(SpringRunner.class) -// @checkstyle:off +@ExtendWith(SpringExtension.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { "spring.cloud.stream.bindings.input.group=" + KafkaBinderActuatorTests.TEST_CONSUMER_GROUP, "spring.cloud.stream.function.bindings.process-in-0=input", "spring.cloud.stream.pollable-source=input"} ) -// @checkstyle:on @DirtiesContext +@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") public class KafkaBinderActuatorTests { static final String TEST_CONSUMER_GROUP = "testGroup-actuatorTests"; - private static final String KAFKA_BROKERS_PROPERTY = "spring.kafka.bootstrap-servers"; - - @ClassRule - public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true); - - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - } - @Autowired private MeterRegistry meterRegistry; @@ -101,7 +81,7 @@ public class KafkaBinderActuatorTests { private KafkaTemplate kafkaTemplate; @Test - public void testKafkaBinderMetricsExposed() { + void testKafkaBinderMetricsExposed() { this.kafkaTemplate.send("input", null, "foo".getBytes()); this.kafkaTemplate.flush(); @@ -111,8 +91,8 @@ public class KafkaBinderActuatorTests { } @Test - @Ignore - public void testKafkaBinderMetricsWhenNoMicrometer() { + @Disabled + void testKafkaBinderMetricsWhenNoMicrometer() { new ApplicationContextRunner().withUserConfiguration(KafkaMetricsTestConfig.class) .withPropertyValues( "spring.cloud.stream.bindings.input.group", KafkaBinderActuatorTests.TEST_CONSUMER_GROUP, diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java index 64b5a79b5..45ea3081f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaBinderExtendedPropertiesTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -25,11 +25,8 @@ import java.util.function.Function; import org.apache.kafka.clients.consumer.Consumer; 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.runner.RunWith; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -45,10 +42,10 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerPro import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; import static org.assertj.core.api.Assertions.assertThat; @@ -56,7 +53,7 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Soby Chacko * @author Gary Russell */ -@RunWith(SpringRunner.class) +@ExtendWith(SpringExtension.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { "spring.cloud.stream.function.definition=process;processCustom", "spring.cloud.stream.function.bindings.process-in-0=standard-in", @@ -75,29 +72,14 @@ import static org.assertj.core.api.Assertions.assertThat; "spring.cloud.stream.kafka.default.consumer.ackEachRecord=true", "spring.cloud.stream.kafka.bindings.custom-in.consumer.ackEachRecord=false" }) @DirtiesContext +@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") public class KafkaBinderExtendedPropertiesTest { - private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; - - @ClassRule - public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true); - - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - } - @Autowired private ConfigurableApplicationContext context; @Test - public void testKafkaBinderExtendedProperties() throws Exception { + void testKafkaBinderExtendedProperties() throws Exception { BinderFactory binderFactory = context.getBeanFactory() .getBean(BinderFactory.class); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java index 660931a1e..ba09b90e7 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-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. @@ -30,12 +30,10 @@ import org.apache.kafka.clients.producer.ProducerInterceptor; 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.runner.RunWith; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer; @@ -43,10 +41,11 @@ import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustom import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; -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.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; import static org.assertj.core.api.Assertions.assertThat; @@ -55,34 +54,24 @@ import static org.assertj.core.api.Assertions.assertThat; * * Based on: https://github.com/spring-projects/spring-kafka/issues/897#issuecomment-466060097 */ -@RunWith(SpringRunner.class) +@ExtendWith(SpringExtension.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = {"spring.cloud.function.definition=process", "spring.cloud.stream.bindings.process-in-0.group=KafkaConfigCustomizationTests.group"}) @DirtiesContext +@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") public class KafkaConfigCustomizationTests { private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; - @ClassRule - public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true); - static final CountDownLatch countDownLatch = new CountDownLatch(2); - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - } + @Autowired + EmbeddedKafkaBroker embeddedKafkaBroker; @Test - public void testBothConsumerAndProducerConfigsCanBeCustomized() throws InterruptedException { + void testBothConsumerAndProducerConfigsCanBeCustomized() throws InterruptedException { Map producerProps = KafkaTestUtils - .producerProps(kafkaEmbedded.getEmbeddedKafka()); + .producerProps(embeddedKafkaBroker); KafkaTemplate template = new KafkaTemplate<>( new DefaultKafkaProducerFactory<>(producerProps)); template.send("process-in-0", "test-foo"); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java index cd5e5386a..1c3d164d5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaNullConverterTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,12 +20,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Consumer; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Ignore; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; @@ -36,12 +33,12 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.KafkaNull; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; import static org.assertj.core.api.Assertions.assertThat; @@ -50,38 +47,23 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Gary Russell * @author Soby Chacko */ -@RunWith(SpringRunner.class) +@ExtendWith(SpringExtension.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { "spring.kafka.consumer.auto-offset-reset=earliest", "spring.cloud.stream.function.bindings.inputListen-in-0=kafkaNullInput"}) @DirtiesContext +@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") public class KafkaNullConverterTest { - private static final String KAFKA_BROKERS_PROPERTY = "spring.kafka.bootstrap-servers"; - @Autowired private ApplicationContext context; @Autowired private KafkaNullConverterTestConfig config; - @ClassRule - public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true); - - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - } - @Test - @Ignore - public void testKafkaNullConverterOutput() throws InterruptedException { + @Disabled + void testKafkaNullConverterOutput() throws InterruptedException { final StreamBridge streamBridge = context.getBean(StreamBridge.class); streamBridge.send("kafkaNullOutput", new GenericMessage<>(KafkaNull.INSTANCE)); @@ -91,7 +73,7 @@ public class KafkaNullConverterTest { } @Test - public void testKafkaNullConverterInput() throws InterruptedException { + void testKafkaNullConverterInput() throws InterruptedException { final MessageChannel kafkaNullInput = context.getBean("kafkaNullInput", MessageChannel.class); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java index 2002ee773..95d6d37fa 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaRetryDlqBinderOrContainerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2021 the original author or authors. + * Copyright 2021-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. @@ -60,7 +60,7 @@ import static org.mockito.Mockito.mock; public class KafkaRetryDlqBinderOrContainerTests { @Test - public void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) { + void retryAndDlqInRightPlace(@Autowired BindingsLifecycleController controller) { Binding retryInBinder = controller.queryState("retryInBinder-in-0"); assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, "lifecycle.retryTemplate")).isNotNull(); assertThat(KafkaTestUtils.getPropertyValue(retryInBinder, diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java index 861a7e152..f8f2afdb7 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/ProducerOnlyTransactionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,18 +18,14 @@ package org.springframework.cloud.stream.binder.kafka.integration; import java.util.Map; -import kafka.server.KafkaConfig; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.IsolationLevel; import org.apache.kafka.common.serialization.ByteArrayDeserializer; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.annotation.Autowired; @@ -42,13 +38,14 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.core.ProducerFactory; -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.kafka.transaction.KafkaTransactionManager; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.annotation.EnableTransactionManagement; import org.springframework.transaction.annotation.Transactional; @@ -63,51 +60,38 @@ import static org.assertj.core.api.Assertions.assertThat; * @since 2.1.4 * */ -@RunWith(SpringRunner.class) +@ExtendWith(SpringExtension.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { "spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx.", "spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=99", "spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all"}) @DirtiesContext +@EmbeddedKafka(topics = "output", controlledShutdown = true, brokerProperties = {"transaction.state.log.replication.factor=1", + "transaction.state.log.min.isr=1"}, bootstrapServersProperty = "spring.kafka.bootstrap-servers") public class ProducerOnlyTransactionTests { - private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; - - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "output") - .brokerProperty(KafkaConfig.TransactionsTopicReplicationFactorProp(), "1") - .brokerProperty(KafkaConfig.TransactionsTopicMinISRProp(), "1"); - @Autowired private Sender sender; @Autowired private ApplicationContext context; - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - embeddedKafka.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - } + @Autowired + EmbeddedKafkaBroker embeddedKafkaBrokera; @Test - public void testProducerTx() { + void testProducerTx() { final StreamBridge streamBridge = context.getBean(StreamBridge.class); this.sender.DoInTransaction(streamBridge); assertThat(this.sender.isInTx()).isTrue(); Map props = KafkaTestUtils.consumerProps("consumeTx", "false", - embeddedKafka.getEmbeddedKafka()); + embeddedKafkaBrokera); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, IsolationLevel.READ_COMMITTED.name().toLowerCase()); Consumer consumer = new KafkaConsumer<>(props); - embeddedKafka.getEmbeddedKafka().consumeFromAllEmbeddedTopics(consumer); + embeddedKafkaBrokera.consumeFromAllEmbeddedTopics(consumer); ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, "output"); assertThat(record.value()).isEqualTo("foo".getBytes()); } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/BaseKafkaBinderTopicPropertiesUpdateTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/BaseKafkaBinderTopicPropertiesUpdateTest.java index 72bec2af8..05801f9a0 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/BaseKafkaBinderTopicPropertiesUpdateTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/BaseKafkaBinderTopicPropertiesUpdateTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2021 the original author or authors. + * Copyright 2018-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -18,23 +18,22 @@ package org.springframework.cloud.stream.binder.kafka.integration.topic.configs; import java.util.function.Function; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.annotation.Bean; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; /** * @author Heiko Does * @author Soby Chacko */ -@RunWith(SpringRunner.class) +@ExtendWith(SpringExtension.class) @SpringBootTest( classes = BaseKafkaBinderTopicPropertiesUpdateTest.TopicAutoConfigsTestConfig.class, webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { @@ -46,23 +45,11 @@ import org.springframework.test.context.junit4.SpringRunner; "spring.cloud.stream.kafka.default.consumer.topic.properties.retention.ms=-1" }) @DirtiesContext +@EmbeddedKafka(bootstrapServersProperty = "spring.kafka.bootstrap-servers") public abstract class BaseKafkaBinderTopicPropertiesUpdateTest { - private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; - - @ClassRule - public static EmbeddedKafkaRule kafkaEmbedded = new EmbeddedKafkaRule(1, true, "standard-in", "standard-out"); - - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - } + @Autowired + protected EmbeddedKafkaBroker embeddedKafka; @EnableAutoConfiguration public static class TopicAutoConfigsTestConfig { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/DisabledKafkaBinderTopicPropertiesUpdateTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/DisabledKafkaBinderTopicPropertiesUpdateTest.java index 89393faf1..a7ca8ec4f 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/DisabledKafkaBinderTopicPropertiesUpdateTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/DisabledKafkaBinderTopicPropertiesUpdateTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,7 +27,7 @@ import org.apache.kafka.clients.admin.Config; import org.apache.kafka.clients.admin.DescribeConfigsResult; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.config.ConfigResource; -import org.junit.Test; +import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat; @@ -37,9 +37,9 @@ import static org.assertj.core.api.Assertions.assertThat; public class DisabledKafkaBinderTopicPropertiesUpdateTest extends BaseKafkaBinderTopicPropertiesUpdateTest { @Test - public void testKafkaBinderShouldNotUpdateTopicConfigurationOnDisabledFeature() throws Exception { + void testKafkaBinderShouldNotUpdateTopicConfigurationOnDisabledFeature() throws Exception { Map adminClientConfig = new HashMap<>(); - adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); + adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); AdminClient adminClient = AdminClient.create(adminClientConfig); ConfigResource standardInConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-in"); ConfigResource standardOutConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-out"); @@ -48,9 +48,9 @@ public class DisabledKafkaBinderTopicPropertiesUpdateTest extends BaseKafkaBinde KafkaFuture> kafkaFuture = describeConfigsResult.all(); Map configResourceConfigMap = kafkaFuture.get(3, TimeUnit.SECONDS); Config standardInTopicConfig = configResourceConfigMap.get(standardInConfigResource); - assertThat(standardInTopicConfig.get("retention.ms").value()).isEqualTo("604800000"); + assertThat(standardInTopicConfig.get("retention.ms").value()).isEqualTo("9001"); Config standardOutTopicConfig = configResourceConfigMap.get(standardOutConfigResource); - assertThat(standardOutTopicConfig.get("retention.ms").value()).isEqualTo("604800000"); + assertThat(standardOutTopicConfig.get("retention.ms").value()).isEqualTo("9001"); } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/KafkaBinderTopicPropertiesUpdateTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/KafkaBinderTopicPropertiesUpdateTest.java index 95b7f9cbc..a1912afd0 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/KafkaBinderTopicPropertiesUpdateTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/topic/configs/KafkaBinderTopicPropertiesUpdateTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -27,7 +27,7 @@ import org.apache.kafka.clients.admin.Config; import org.apache.kafka.clients.admin.DescribeConfigsResult; import org.apache.kafka.common.KafkaFuture; import org.apache.kafka.common.config.ConfigResource; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.test.context.TestPropertySource; @@ -40,9 +40,9 @@ import static org.assertj.core.api.Assertions.assertThat; public class KafkaBinderTopicPropertiesUpdateTest extends BaseKafkaBinderTopicPropertiesUpdateTest { @Test - public void testKafkaBinderUpdateTopicConfiguration() throws Exception { + void testKafkaBinderUpdateTopicConfiguration() throws Exception { Map adminClientConfig = new HashMap<>(); - adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaEmbedded.getEmbeddedKafka().getBrokersAsString()); + adminClientConfig.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString()); AdminClient adminClient = AdminClient.create(adminClientConfig); ConfigResource standardInConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-in"); ConfigResource standardOutConfigResource = new ConfigResource(ConfigResource.Type.TOPIC, "standard-out"); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java index dbf30a093..15ce8b566 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration2/ConsumerProducerTransactionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -23,12 +23,8 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.function.Function; -import kafka.server.KafkaConfig; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.ApplicationRunner; @@ -43,12 +39,12 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.DefaultAfterRollbackProcessor; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.kafka.transaction.KafkaAwareTransactionManager; import org.springframework.messaging.MessageChannel; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringExtension; import org.springframework.util.backoff.FixedBackOff; import static org.assertj.core.api.Assertions.assertThat; @@ -61,7 +57,7 @@ import static org.mockito.Mockito.mock; * @since 3.0 * */ -@RunWith(SpringRunner.class) +@ExtendWith(SpringExtension.class) @SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { "spring.kafka.consumer.properties.isolation.level=read_committed", "spring.kafka.consumer.enable-auto-commit=false", @@ -81,35 +77,16 @@ import static org.mockito.Mockito.mock; "spring.cloud.stream.kafka.binder.transaction.producer.configuration.retries=99", "spring.cloud.stream.kafka.binder.transaction.producer.configuration.acks=all"}) @DirtiesContext +@EmbeddedKafka(topics = "consumer.producer.txOut", controlledShutdown = true, brokerProperties = {"transaction.state.log.replication.factor=1", + "transaction.state.log.min.isr=1"}, bootstrapServersProperty = "spring.kafka.bootstrap-servers") public class ConsumerProducerTransactionTests { - private static final String KAFKA_BROKERS_PROPERTY = "spring.cloud.stream.kafka.binder.brokers"; - - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, "consumer.producer.txOut") - .brokerProperty(KafkaConfig.TransactionsTopicReplicationFactorProp(), "1") - .brokerProperty(KafkaConfig.TransactionsTopicMinISRProp(), "1"); - @Autowired private Config config; @Autowired private ApplicationContext context; - @BeforeClass - public static void setup() { - System.setProperty(KAFKA_BROKERS_PROPERTY, - embeddedKafka.getEmbeddedKafka().getBrokersAsString()); - System.setProperty("spring.kafka.bootstrap-servers", - embeddedKafka.getEmbeddedKafka().getBrokersAsString()); - } - - @AfterClass - public static void clean() { - System.clearProperty(KAFKA_BROKERS_PROPERTY); - System.clearProperty("spring.kafka.bootstrap-servers"); - } - @Test public void testProducerRunsInConsumerTransaction() throws InterruptedException { assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue(); @@ -117,7 +94,7 @@ public class ConsumerProducerTransactionTests { } @Test - public void externalTM() { + void externalTM() { assertThat(this.config.input2Container.getContainerProperties().getTransactionManager()) .isSameAs(this.config.tm); final MessageChannel output2 = context.getBean("output2", MessageChannel.class);