From 840933d164a6a3c243cc4e2eabf48737c6d7baff Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 15 Mar 2022 18:45:05 -0400 Subject: [PATCH] JUnit 5 migration for Kafka Streams binder tests --- .../KafkaStreamsEventTypeRoutingTests.java | 27 ++++++-------- .../KafkaStreamsFunctionCompositionTests.java | 37 +++++++++---------- ...reamsInteractiveQueryIntegrationTests.java | 28 ++++++-------- .../MultipleFunctionsInSameAppTests.java | 27 ++++++-------- .../KafkaStreamsBinderBootstrapTest.java | 33 +++++++++-------- .../KafkaStreamsBinderJaasInitTests.java | 25 +++++++------ ...sBinderWordCountBranchesFunctionTests.java | 25 ++++++------- ...kaStreamsBinderWordCountFunctionTests.java | 35 ++++++++---------- .../KafkaStreamsComponentBeansTests.java | 35 ++++++++---------- .../KafkaStreamsFunctionStateStoreTests.java | 17 ++++----- .../function/KafkaStreamsRetryTests.java | 20 +++++----- .../function/SerdesProvidedAsBeansTests.java | 16 ++++---- .../StreamToGlobalKTableFunctionTests.java | 19 ++++------ .../StreamToTableJoinFunctionTests.java | 21 +++++------ .../DlqDestinationResolverTests.java | 19 ++++------ ...treamsBinderDestinationIsPatternTests.java | 26 ++++++------- ...afkaStreamsBinderHealthIndicatorTests.java | 24 +++++------- ...aStreamsBinderMultipleInputTopicsTest.java | 26 ++++++------- ...rPojoInputAndPrimitiveTypeOutputTests.java | 26 ++++++------- .../KafkaStreamsBinderTombstoneTests.java | 26 ++++++------- ...afkaStreamsStateStoreIntegrationTests.java | 20 ++++------ ...PojoInputStringOutputIntegrationTests.java | 26 ++++++------- .../streams/serde/CollectionSerdeTest.java | 4 +- 23 files changed, 248 insertions(+), 314 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java index 603d43247..f7c9fd8a3 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 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. @@ -34,10 +34,9 @@ import org.apache.kafka.common.header.internals.RecordHeaders; import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -50,25 +49,23 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.Assert; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka(topics = "foo-2") public class KafkaStreamsEventTypeRoutingTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "foo-1", "foo-2"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; private static CountDownLatch LATCH = new CountDownLatch(3); - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false", embeddedKafka); @@ -80,14 +77,14 @@ public class KafkaStreamsEventTypeRoutingTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "foo-2"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } //See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1003 for more context on this test. @Test - public void testRoutingWorksBasedOnEventTypes() { + void testRoutingWorksBasedOnEventTypes() { SpringApplication app = new SpringApplication(EventTypeRoutingTestConfig.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -157,7 +154,7 @@ public class KafkaStreamsEventTypeRoutingTests { } @Test - public void testRoutingWorksBasedOnEventTypesConsumer() throws Exception { + void testRoutingWorksBasedOnEventTypesConsumer() throws Exception { SpringApplication app = new SpringApplication(EventTypeRoutingTestConfig.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java index cebe3121a..8e889767b 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.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. @@ -32,10 +32,9 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.streams.kstream.ForeachAction; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -46,19 +45,17 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.Assert; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka(topics = {"fooFuncanotherFooFunc-out-0", "bar"}) public class KafkaStreamsFunctionCompositionTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "fooFuncanotherFooFunc-out-0", "bar"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; @@ -66,7 +63,7 @@ public class KafkaStreamsFunctionCompositionTests { private static final CountDownLatch countDownLatch2 = new CountDownLatch(1); private static final CountDownLatch countDownLatch3 = new CountDownLatch(2); - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("fn-composition-group", "false", embeddedKafka); @@ -77,13 +74,13 @@ public class KafkaStreamsFunctionCompositionTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "fooFuncanotherFooFunc-out-0", "bar"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testBasicFunctionCompositionWithDefaultDestination() throws InterruptedException { + void testBasicFunctionCompositionWithDefaultDestination() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig1.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -117,7 +114,7 @@ public class KafkaStreamsFunctionCompositionTests { } @Test - public void testBasicFunctionCompositionWithDestinaion() throws InterruptedException { + void testBasicFunctionCompositionWithDestinaion() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig1.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -153,7 +150,7 @@ public class KafkaStreamsFunctionCompositionTests { } @Test - public void testFunctionToConsumerComposition() throws InterruptedException { + void testFunctionToConsumerComposition() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig2.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -183,7 +180,7 @@ public class KafkaStreamsFunctionCompositionTests { } @Test - public void testBiFunctionToConsumerComposition() throws InterruptedException { + void testBiFunctionToConsumerComposition() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig3.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -217,7 +214,7 @@ public class KafkaStreamsFunctionCompositionTests { } @Test - public void testChainedFunctionsAsComposed() throws InterruptedException { + void testChainedFunctionsAsComposed() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig4.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -260,7 +257,7 @@ public class KafkaStreamsFunctionCompositionTests { } @Test - public void testFirstFunctionCurriedThenComposeWithOtherFunctions() throws InterruptedException { + void testFirstFunctionCurriedThenComposeWithOtherFunctions() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig5.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -305,7 +302,7 @@ public class KafkaStreamsFunctionCompositionTests { } @Test - public void testFunctionToConsumerCompositionWithFunctionProducesKTable() throws InterruptedException { + void testFunctionToConsumerCompositionWithFunctionProducesKTable() throws InterruptedException { SpringApplication app = new SpringApplication(FunctionCompositionConfig6.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java index 4eab47fbe..023a5a476 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java @@ -39,10 +39,9 @@ import org.apache.kafka.streams.state.HostInfo; import org.apache.kafka.streams.state.QueryableStoreType; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; -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.Mockito; import org.springframework.boot.SpringApplication; @@ -59,7 +58,8 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -71,18 +71,14 @@ import static org.mockito.internal.verification.VerificationModeFactory.times; * @author Gary Russell * @author Nico Pommerening */ +@EmbeddedKafka(topics = "counts-id") public class KafkaStreamsInteractiveQueryIntegrationTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts-id"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); @@ -93,13 +89,13 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testStateStoreRetrievalRetry() { + void testStateStoreRetrievalRetry() { StreamsBuilderFactoryBean mock = Mockito.mock(StreamsBuilderFactoryBean.class); KafkaStreams mockKafkaStreams = Mockito.mock(KafkaStreams.class); @@ -128,7 +124,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } @Test - public void testStateStoreRetrievalRetryForHostInfoService() { + void testStateStoreRetrievalRetryForHostInfoService() { StreamsBuilderFactoryBean mock = Mockito.mock(StreamsBuilderFactoryBean.class); KafkaStreams mockKafkaStreams = Mockito.mock(KafkaStreams.class); Mockito.when(mock.getKafkaStreams()).thenReturn(mockKafkaStreams); @@ -157,7 +153,7 @@ public class KafkaStreamsInteractiveQueryIntegrationTests { } @Test - public void testKstreamBinderWithPojoInputAndStringOuput() { + void testKstreamBinderWithPojoInputAndStringOuput() { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java index b64d4c406..d9679f926 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -30,10 +30,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.KStream; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -46,26 +45,24 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka(topics = {"coffee", "electronics"}) public class MultipleFunctionsInSameAppTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "coffee", "electronics"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; private static CountDownLatch countDownLatch = new CountDownLatch(2); - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("purchase-groups", "false", embeddedKafka); @@ -75,14 +72,14 @@ public class MultipleFunctionsInSameAppTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "coffee", "electronics"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test @SuppressWarnings("unchecked") - public void testMultiFunctionsInSameApp() throws InterruptedException { + void testMultiFunctionsInSameApp() throws InterruptedException { SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -140,7 +137,7 @@ public class MultipleFunctionsInSameAppTests { } @Test - public void testMultiFunctionsInSameAppWithMultiBinders() throws Exception { + void testMultiFunctionsInSameAppWithMultiBinders() throws Exception { SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java index c03887691..2ffc8ab6a 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.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. @@ -27,9 +27,8 @@ import org.apache.kafka.common.security.JaasUtils; import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; -import org.junit.Before; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -38,7 +37,9 @@ import org.springframework.cloud.stream.binder.kafka.streams.KeyValueSerdeResolv import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.config.StreamsBuilderFactoryBean; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; @@ -46,18 +47,18 @@ import static org.assertj.core.api.AssertionsForClassTypes.assertThat; * @author Soby Chacko * @author Eduard Domínguez */ +@EmbeddedKafka public class KafkaStreamsBinderBootstrapTest { - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - @Before + @BeforeEach public void before() { System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); } @Test - public void testKStreamBinderWithCustomEnvironmentCanStart() { + void testKStreamBinderWithCustomEnvironmentCanStart() { ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder( SimpleKafkaStreamsApplication.class).web(WebApplicationType.NONE).run( "--spring.cloud.function.definition=input1;input2;input3", @@ -72,25 +73,25 @@ public class KafkaStreamsBinderBootstrapTest { "--spring.cloud.stream.binders.kstreamBinder.type=kstream", "--spring.cloud.stream.binders.kstreamBinder.environment" + ".spring.cloud.stream.kafka.streams.binder.brokers" - + "=" + embeddedKafka.getEmbeddedKafka().getBrokersAsString(), + + "=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.bindings.input2-in-0.destination=bar", "--spring.cloud.stream.bindings.input2-in-0.binder=ktableBinder", "--spring.cloud.stream.binders.ktableBinder.type=ktable", "--spring.cloud.stream.binders.ktableBinder.environment" + ".spring.cloud.stream.kafka.streams.binder.brokers" - + "=" + embeddedKafka.getEmbeddedKafka().getBrokersAsString(), + + "=" + embeddedKafka.getBrokersAsString(), "--spring.cloud.stream.bindings.input3-in-0.destination=foobar", "--spring.cloud.stream.bindings.input3-in-0.binder=globalktableBinder", "--spring.cloud.stream.binders.globalktableBinder.type=globalktable", "--spring.cloud.stream.binders.globalktableBinder.environment" + ".spring.cloud.stream.kafka.streams.binder.brokers" - + "=" + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); + + "=" + embeddedKafka.getBrokersAsString()); applicationContext.close(); } @Test - public void testKafkaStreamsBinderWithStandardConfigurationCanStart() { + void testKafkaStreamsBinderWithStandardConfigurationCanStart() { ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder( SimpleKafkaStreamsApplication.class).web(WebApplicationType.NONE).run( "--spring.cloud.function.definition=input1;input2;input3", @@ -101,14 +102,14 @@ public class KafkaStreamsBinderBootstrapTest { "--spring.cloud.stream.kafka.streams.bindings.input3-in-0.consumer.application-id" + "=testKafkaStreamsBinderWithStandardConfigurationCanStart-foobar", "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); + + embeddedKafka.getBrokersAsString()); applicationContext.close(); } @Test @SuppressWarnings("unchecked") - public void testStreamConfigGlobalProperties_GH1149() { + void testStreamConfigGlobalProperties_GH1149() { ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder( SimpleKafkaStreamsApplication.class).web(WebApplicationType.NONE).run( "--spring.cloud.function.definition=input1;input2;input3", @@ -120,7 +121,7 @@ public class KafkaStreamsBinderBootstrapTest { "--spring.cloud.stream.kafka.streams.bindings.input3-in-0.consumer.application-id" + "=testKafkaStreamsBinderWithStandardConfigurationCanStart-foobar", "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); + + embeddedKafka.getBrokersAsString()); Map streamConfigGlobalProperties = applicationContext .getBean("streamConfigGlobalProperties", Map.class); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderJaasInitTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderJaasInitTests.java index 7f52699a8..83d181ed1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderJaasInitTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderJaasInitTests.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. @@ -22,34 +22,35 @@ import javax.security.auth.login.AppConfigurationEntry; import org.apache.kafka.common.security.JaasUtils; import org.apache.kafka.streams.kstream.KStream; -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.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.EmbeddedKafkaBroker; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka public class KafkaStreamsBinderJaasInitTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafka = new EmbeddedKafkaRule(1, true, 10); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static String JAVA_LOGIN_CONFIG_PARAM_VALUE; - @BeforeClass + @BeforeAll public static void beforeAll() { JAVA_LOGIN_CONFIG_PARAM_VALUE = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); } - @AfterClass + @AfterAll public static void afterAll() { if (JAVA_LOGIN_CONFIG_PARAM_VALUE != null) { System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, JAVA_LOGIN_CONFIG_PARAM_VALUE); @@ -57,7 +58,7 @@ public class KafkaStreamsBinderJaasInitTests { } @Test - public void testKafkaStreamsBinderJaasInitialization() { + void testKafkaStreamsBinderJaasInitialization() { ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder( KafkaStreamsBinderJaasInitTestsApplication.class).web(WebApplicationType.NONE).run( "--spring.cloud.function.definition=foo", @@ -67,7 +68,7 @@ public class KafkaStreamsBinderJaasInitTests { "--spring.cloud.stream.kafka.streams.binder.jaas.options.username=foo", "--spring.cloud.stream.kafka.streams.binder.jaas.options.password=bar", "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getEmbeddedKafka().getBrokersAsString()); + + embeddedKafka.getBrokersAsString()); javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration .getConfiguration(); final AppConfigurationEntry[] kafkaConfiguration = configuration diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java index 1f1d443ef..68ade8680 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.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. @@ -30,10 +30,9 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.Predicate; import org.apache.kafka.streams.kstream.TimeWindows; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -44,7 +43,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -52,17 +52,14 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Soby Chacko */ +@EmbeddedKafka(topics = {"counts", "foo", "bar"}) public class KafkaStreamsBinderWordCountBranchesFunctionTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts", "foo", "bar"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("groupx", "false", embeddedKafka); @@ -72,13 +69,13 @@ public class KafkaStreamsBinderWordCountBranchesFunctionTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "counts", "foo", "bar"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { + void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index 65276e189..1a61ddf31 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.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. @@ -38,10 +38,9 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; import org.apache.kafka.streams.processor.StreamPartitioner; -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.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; @@ -65,25 +64,23 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.Assert; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka(topics = {"counts", "counts-1", "counts-2", "counts-5", "counts-6"}) public class KafkaStreamsBinderWordCountFunctionTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts", "counts-1", "counts-2", "counts-5", "counts-6"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; private final static CountDownLatch LATCH = new CountDownLatch(1); - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); @@ -94,14 +91,14 @@ public class KafkaStreamsBinderWordCountFunctionTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "counts", "counts-1", "counts-2", "counts-5", "counts-6"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test @SuppressWarnings("unchecked") - public void testBasicKStreamTopologyExecution() throws Exception { + void testBasicKStreamTopologyExecution() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -178,7 +175,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { } @Test - public void testKstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer() { + void testKstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer() { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -199,7 +196,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { } @Test - public void testKstreamWordCountFunctionWithCustomProducerStreamPartitioner() throws Exception { + void testKstreamWordCountFunctionWithCustomProducerStreamPartitioner() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -239,7 +236,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { } @Test - public void testKstreamBinderAutoStartup() throws Exception { + void testKstreamBinderAutoStartup() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -261,7 +258,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { } @Test - public void testKstreamIndividualBindingAutoStartup() throws Exception { + void testKstreamIndividualBindingAutoStartup() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -286,7 +283,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { // The following test verifies the fixes made for this issue: // https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/774 @Test - public void testOutboundNullValueIsHandledGracefully() + void testOutboundNullValueIsHandledGracefully() throws Exception { SpringApplication app = new SpringApplication(OutboundNullApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java index cc4a5259d..c80a657b1 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsComponentBeansTests.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. @@ -29,10 +29,9 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -42,7 +41,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.stereotype.Component; import org.springframework.util.Assert; @@ -52,13 +52,10 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Soby Chacko */ +@EmbeddedKafka(topics = {"testFunctionComponent-out", "testBiFunctionComponent-out", "testCurriedFunctionWithFunctionTerminal-out"}) public class KafkaStreamsComponentBeansTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "testFunctionComponent-out", "testBiFunctionComponent-out", "testCurriedFunctionWithFunctionTerminal-out"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer1; private static Consumer consumer2; @@ -68,7 +65,7 @@ public class KafkaStreamsComponentBeansTests { private final static CountDownLatch LATCH_2 = new CountDownLatch(2); private final static CountDownLatch LATCH_3 = new CountDownLatch(3); - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); @@ -95,7 +92,7 @@ public class KafkaStreamsComponentBeansTests { embeddedKafka.consumeFromEmbeddedTopics(consumer3, "testCurriedFunctionWithFunctionTerminal-out"); } - @AfterClass + @AfterAll public static void tearDown() { consumer1.close(); consumer2.close(); @@ -103,7 +100,7 @@ public class KafkaStreamsComponentBeansTests { } @Test - public void testFunctionComponent() { + void testFunctionComponent() { SpringApplication app = new SpringApplication(FunctionAsComponent.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run( @@ -129,7 +126,7 @@ public class KafkaStreamsComponentBeansTests { } @Test - public void testConsumerComponent() throws Exception { + void testConsumerComponent() throws Exception { SpringApplication app = new SpringApplication(ConsumerAsComponent.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext context = app.run( @@ -153,7 +150,7 @@ public class KafkaStreamsComponentBeansTests { } @Test - public void testBiFunctionComponent() { + void testBiFunctionComponent() { SpringApplication app = new SpringApplication(BiFunctionAsComponent.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run( @@ -183,7 +180,7 @@ public class KafkaStreamsComponentBeansTests { } @Test - public void testBiConsumerComponent() throws Exception { + void testBiConsumerComponent() throws Exception { SpringApplication app = new SpringApplication(BiConsumerAsComponent.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext context = app.run( @@ -210,7 +207,7 @@ public class KafkaStreamsComponentBeansTests { } @Test - public void testCurriedFunctionWithConsumerTerminal() throws Exception { + void testCurriedFunctionWithConsumerTerminal() throws Exception { SpringApplication app = new SpringApplication(CurriedFunctionWithConsumerTerminal.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext context = app.run( @@ -240,7 +237,7 @@ public class KafkaStreamsComponentBeansTests { } @Test - public void testCurriedFunctionWithFunctionTerminal() { + void testCurriedFunctionWithFunctionTerminal() { SpringApplication app = new SpringApplication(CurriedFunctionWithFunctionTerminal.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext context = app.run( diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java index 36747368a..3d70a9970 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -29,8 +29,7 @@ import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.streams.state.WindowStore; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -40,21 +39,19 @@ import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka public class KafkaStreamsFunctionStateStoreTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); @Test - public void testKafkaStreamsFuncionWithMultipleStateStores() throws Exception { + void testKafkaStreamsFuncionWithMultipleStateStores() throws Exception { SpringApplication app = new SpringApplication(StateStoreTestApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java index 285822314..3e3c44f25 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.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. @@ -26,8 +26,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorContext; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.annotation.Qualifier; @@ -41,7 +40,8 @@ import org.springframework.context.annotation.Lazy; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.retry.RetryPolicy; import org.springframework.retry.backoff.FixedBackOffPolicy; @@ -52,18 +52,16 @@ import org.springframework.util.Assert; import static org.assertj.core.api.AssertionsForClassTypes.assertThatThrownBy; import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat; +@EmbeddedKafka public class KafkaStreamsRetryTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true); - - private static final EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private final static CountDownLatch LATCH1 = new CountDownLatch(2); private final static CountDownLatch LATCH2 = new CountDownLatch(4); @Test - public void testRetryTemplatePerBindingOnKStream() throws Exception { + void testRetryTemplatePerBindingOnKStream() throws Exception { SpringApplication app = new SpringApplication(RetryTemplatePerConsumerBindingApp.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -85,7 +83,7 @@ public class KafkaStreamsRetryTests { } @Test - public void testRetryTemplateOnTableTypes() throws Exception { + void testRetryTemplateOnTableTypes() throws Exception { SpringApplication app = new SpringApplication(RetryTemplatePerConsumerBindingApp.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -102,7 +100,7 @@ public class KafkaStreamsRetryTests { } @Test - public void testRetryTemplateBeanProvidedByTheApp() throws Exception { + void testRetryTemplateBeanProvidedByTheApp() throws Exception { SpringApplication app = new SpringApplication(CustomRetryTemplateApp.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java index f5893ce84..ecda1364e 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -24,8 +24,7 @@ import org.apache.kafka.common.serialization.Deserializer; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.kstream.KStream; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -41,18 +40,17 @@ import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.core.ResolvableType; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.util.Assert; +@EmbeddedKafka public class SerdesProvidedAsBeansTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); @Test - public void testKstreamWordCountFunction() throws NoSuchMethodException { + void testKstreamWordCountFunction() throws NoSuchMethodException { SpringApplication app = new SpringApplication(SerdeProvidedAsBeanApp.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java index b7eb33172..4f55d3997 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -35,8 +35,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.processor.TimestampExtractor; import org.apache.kafka.streams.processor.WallclockTimestampExtractor; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -57,23 +56,21 @@ import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka(topics = "enriched-order") public class StreamToGlobalKTableFunctionTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "enriched-order"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; @Test - public void testStreamToGlobalKTable() throws Exception { + void testStreamToGlobalKTable() throws Exception { SpringApplication app = new SpringApplication(OrderEnricherApplication.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext context = app.run("--server.port=0", @@ -225,7 +222,7 @@ public class StreamToGlobalKTableFunctionTests { } @Test - public void testTimeExtractor() throws Exception { + void testTimeExtractor() throws Exception { SpringApplication app = new SpringApplication(OrderEnricherApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java index 6f269fa11..6fba778d8 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -46,8 +46,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.StreamJoined; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -66,22 +65,20 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.Assert; import static org.assertj.core.api.Assertions.assertThat; +@EmbeddedKafka(topics = "output-topic-1") public class StreamToTableJoinFunctionTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, - true, "output-topic-1", "output-topic-2", "user-clicks-2", "user-regions-2"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); @Test - public void testStreamToTable() { + void testStreamToTable() { SpringApplication app = new SpringApplication(CountClicksPerRegionApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -99,7 +96,7 @@ public class StreamToTableJoinFunctionTests { } @Test - public void testStreamToTableBiFunction() { + void testStreamToTableBiFunction() { SpringApplication app = new SpringApplication(BiFunctionCountClicksPerRegionApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -117,7 +114,7 @@ public class StreamToTableJoinFunctionTests { } @Test - public void testStreamToTableBiConsumer() throws Exception { + void testStreamToTableBiConsumer() throws Exception { SpringApplication app = new SpringApplication(BiConsumerApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java index e45bc639a..5b916ecc0 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 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,8 +30,7 @@ import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -45,7 +44,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -53,18 +53,13 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Soby Chacko */ +@EmbeddedKafka(topics = {"topic1-dlq", "topic2-dlq"}) public class DlqDestinationResolverTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "topic1-dlq", - "topic2-dlq"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); @Test - public void testDlqDestinationResolverWorks() throws Exception { + void testDlqDestinationResolverWorks() throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java index 47004a293..9ea6a9cdc 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -21,10 +21,9 @@ import java.util.function.Function; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.streams.kstream.KStream; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -35,7 +34,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -43,18 +43,14 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Michael Stoettinger */ +@EmbeddedKafka(topics = {"in.1", "in.2", "out"}) public class KafkaStreamsBinderDestinationIsPatternTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "in.1", "in.2", "out"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static org.apache.kafka.clients.consumer.Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("group", "true", embeddedKafka); @@ -64,13 +60,13 @@ public class KafkaStreamsBinderDestinationIsPatternTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "out"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void test() { + void test() { SpringApplication app = new SpringApplication(ConsumingApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java index e1a2f1f3f..69f32d64a 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -30,9 +30,8 @@ import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler; import org.apache.kafka.streams.kstream.KStream; import org.assertj.core.util.Lists; import org.junit.Assert; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -50,7 +49,8 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; @@ -60,22 +60,18 @@ import static org.assertj.core.api.Assertions.assertThat; /** * @author Arnaud Jardiné */ +@EmbeddedKafka(topics = {"out", "out2"}) public class KafkaStreamsBinderHealthIndicatorTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "out", "out2"); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); - - @BeforeClass + @BeforeAll public static void setUp() { System.setProperty("logging.level.org.apache.kafka", "OFF"); } @Test - public void healthIndicatorUpTest() throws Exception { + void healthIndicatorUpTest() throws Exception { try (ConfigurableApplicationContext context = singleStream("ApplicationHealthTest-xyz")) { receive(context, Lists.newArrayList(new ProducerRecord<>("in", "{\"id\":\"123\"}"), @@ -85,7 +81,7 @@ public class KafkaStreamsBinderHealthIndicatorTests { } @Test - public void healthIndicatorUpMultipleCallsTest() throws Exception { + void healthIndicatorUpMultipleCallsTest() throws Exception { try (ConfigurableApplicationContext context = singleStream("ApplicationHealthTest-xyz")) { int callsToPerform = 5; for (int i = 0; i < callsToPerform; i++) { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java index 2b841e77f..6ea7ff7e5 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-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,10 +30,9 @@ import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -45,7 +44,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -61,18 +61,14 @@ import static org.assertj.core.api.Assertions.assertThat; * * @author Sarath Shyam */ +@EmbeddedKafka(topics = "counts") public class KafkaStreamsBinderMultipleInputTopicsTest { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); @@ -83,13 +79,13 @@ public class KafkaStreamsBinderMultipleInputTopicsTest { embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { + void testKstreamWordCountWithStringInputAndPojoOuput() throws Exception { SpringApplication app = new SpringApplication( WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java index d95658cee..7ac8bf127 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-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. @@ -29,10 +29,9 @@ import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -44,7 +43,8 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -53,18 +53,14 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Soby Chacko * @author Gary Russell */ +@EmbeddedKafka(topics = "counts-id") public class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts-id"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); @@ -76,13 +72,13 @@ public class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { + void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderTombstoneTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderTombstoneTests.java index f8e118076..9b48d3dd6 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderTombstoneTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderTombstoneTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-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. @@ -35,10 +35,9 @@ import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -52,7 +51,8 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -62,18 +62,14 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Soby Chacko * @author Gary Russell */ +@EmbeddedKafka(topics = "counts-1") public class KafkaStreamsBinderTombstoneTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts-1"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); @@ -84,13 +80,13 @@ public class KafkaStreamsBinderTombstoneTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "counts-1"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testSendToTombstone() + void testSendToTombstone() throws Exception { SpringApplication app = new SpringApplication( WordCountProcessorApplication.class); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java index 22f0bf06d..efddbad18 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.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. @@ -28,8 +28,7 @@ import org.apache.kafka.streams.processor.ProcessorContext; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.streams.state.WindowStore; -import org.junit.ClassRule; -import org.junit.Test; +import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -39,7 +38,8 @@ import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static junit.framework.TestCase.fail; @@ -49,17 +49,13 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Lei Chen * @author Soby Chacko */ +@EmbeddedKafka public class KafkaStreamsStateStoreIntegrationTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts-id"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); @Test - public void testKstreamStateStore() throws Exception { + void testKstreamStateStore() throws Exception { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", @@ -88,7 +84,7 @@ public class KafkaStreamsStateStoreIntegrationTests { } @Test - public void testSameStateStoreIsCreatedOnlyOnceWhenMultipleInputBindingsArePresent() throws Exception { + void testSameStateStoreIsCreatedOnlyOnceWhenMultipleInputBindingsArePresent() throws Exception { SpringApplication app = new SpringApplication(ProductCountApplicationWithMultipleInputBindings.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java index 130db7be4..9fc2f5a3d 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2018 the original author or authors. + * Copyright 2017-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -28,10 +28,9 @@ import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; -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.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; @@ -46,7 +45,8 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.serializer.JsonSerde; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -56,18 +56,14 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Soby Chacko * @author Gary Russell */ +@EmbeddedKafka(topics = "counts-id") public class KafkastreamsBinderPojoInputStringOutputIntegrationTests { - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "counts-id"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); + private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); @@ -78,13 +74,13 @@ public class KafkastreamsBinderPojoInputStringOutputIntegrationTests { embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "counts-id"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); } @Test - public void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { + void testKstreamBinderWithPojoInputAndStringOuput() throws Exception { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CollectionSerdeTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CollectionSerdeTest.java index 0b060d7a5..c31715bcb 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CollectionSerdeTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/serde/CollectionSerdeTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 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. @@ -21,7 +21,7 @@ import java.util.Collection; import java.util.Iterator; import java.util.List; -import org.junit.Test; +import org.junit.jupiter.api.Test; import static org.assertj.core.api.Assertions.assertThat;