From d534e840dcd488de5bdd5b30bd539c165a728991 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 1 May 2024 11:52:44 -0400 Subject: [PATCH] GH-2942: EmbeddedKafka usage improvements in Kafka Streams binder * Currently, EmbeddedKafka is initialized as part of the class initialization. This is preventing individual JUnit tests from being executed from an IDE (IntelliJ, for example). If we move this initialization to the JUnit `BeforeAll` method, then that seems to be working. Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2942 --- .../EventTypeRoutingWithInferredSerdeTests.java | 7 ++++--- ...InteractiveQueryServiceMultiStateStoreTests.java | 11 +++++++++-- ...aStreamsBinderEnvironmentPostProcessorTests.java | 10 ++++++++-- .../streams/KafkaStreamsEventTypeRoutingTests.java | 9 +++++---- .../KafkaStreamsFunctionCompositionTests.java | 7 ++++--- ...afkaStreamsInteractiveQueryIntegrationTests.java | 5 +++-- .../streams/MultipleFunctionsInSameAppTests.java | 9 +++++---- .../kafka/streams/SerdeResolverUtilsTests.java | 8 ++++---- .../bootstrap/KafkaStreamsBinderBootstrapTest.java | 5 +++-- .../bootstrap/KafkaStreamsBinderJaasInitTests.java | 5 +++-- ...StreamsBinderWordCountBranchesFunctionTests.java | 5 +++-- .../KafkaStreamsBinderWordCountFunctionTests.java | 5 +++-- .../KafkaStreamsFunctionStateStoreTests.java | 10 ++++++++-- .../streams/function/KafkaStreamsRetryTests.java | 13 ++++++++++--- .../function/SerdesProvidedAsBeansTests.java | 10 ++++++++-- .../function/StreamToGlobalKTableFunctionTests.java | 12 +++++++++--- .../function/StreamToTableJoinFunctionTests.java | 10 ++++++++-- .../integration/DlqDestinationResolverTests.java | 10 ++++++++-- .../streams/integration/DltAwareProcessorTests.java | 7 ++++--- ...KafkaStreamsBinderDestinationIsPatternTests.java | 5 +++-- .../KafkaStreamsBinderHealthIndicatorTests.java | 5 +++-- .../KafkaStreamsBinderMultipleInputTopicsTest.java | 7 ++++--- ...sBinderPojoInputAndPrimitiveTypeOutputTests.java | 5 +++-- .../KafkaStreamsBinderTombstoneTests.java | 5 +++-- .../KafkaStreamsStateStoreIntegrationTests.java | 10 ++++++++-- ...BinderPojoInputStringOutputIntegrationTests.java | 5 +++-- 26 files changed, 136 insertions(+), 64 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java index df04b02ae..25869411b 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-2024 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. @@ -59,12 +59,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "foo-2") class EventTypeRoutingWithInferredSerdeTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; + private static EmbeddedKafkaBroker embeddedKafka; + @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java index 30363d030..9d51e5647 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2023 the original author or authors. + * Copyright 2022-2024 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,6 +30,7 @@ import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.mockito.Mockito; import org.slf4j.Logger; @@ -63,9 +64,15 @@ import static org.mockito.Mockito.when; class InteractiveQueryServiceMultiStateStoreTests { private static final String STORE_1_NAME = "store1"; + private static final String STORE_2_NAME = "store2"; - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void stateStoreAvailableOnProperAppWhenAppServerPropertySet() { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java index 4a4beab5d..de994ecc7 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2023 the original author or authors. + * Copyright 2022-2024 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. @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams; import java.util.function.Consumer; import org.apache.kafka.streams.kstream.KStream; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.WebApplicationType; @@ -40,7 +41,12 @@ import static org.assertj.core.api.AssertionsForClassTypes.assertThat; @EmbeddedKafka class KafkaStreamsBinderEnvironmentPostProcessorTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void defaultIneligibleFunctionIsSet() { 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 a6309c736..e6b83b1b3 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-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -59,14 +59,15 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "foo-2") class KafkaStreamsEventTypeRoutingTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; - private static CountDownLatch LATCH = new CountDownLatch(3); + private static final CountDownLatch LATCH = new CountDownLatch(3); + + private static EmbeddedKafkaBroker embeddedKafka; @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 a6015c35c..822ce83d5 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-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -55,16 +55,17 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"fooFuncanotherFooFunc-out-0", "bar"}) class KafkaStreamsFunctionCompositionTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; private static final CountDownLatch countDownLatch1 = new CountDownLatch(1); private static final CountDownLatch countDownLatch2 = new CountDownLatch(1); private static final CountDownLatch countDownLatch3 = new CountDownLatch(2); + private static EmbeddedKafkaBroker embeddedKafka; + @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("fn-composition-group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 b35d38198..0d86a5220 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 @@ -74,12 +74,13 @@ import static org.mockito.internal.verification.VerificationModeFactory.times; @EmbeddedKafka(topics = "counts-id") class KafkaStreamsInteractiveQueryIntegrationTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; + private static EmbeddedKafkaBroker embeddedKafka; + @BeforeAll public static void setUp() throws Exception { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 7f8c926cb..a284e3ff3 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-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -56,14 +56,15 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"coffee", "electronics"}) class MultipleFunctionsInSameAppTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; - private static CountDownLatch countDownLatch = new CountDownLatch(2); + private static final CountDownLatch countDownLatch = new CountDownLatch(2); + + private static EmbeddedKafkaBroker embeddedKafka; @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("purchase-groups", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java index dc44d4f12..bc6d92817 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java @@ -49,16 +49,16 @@ import static org.mockito.Mockito.mock; * * @author Chris Bono */ -@SuppressWarnings({ "rawtypes", "NewClassNamingConvention", "unchecked" }) +@SuppressWarnings({ "rawtypes", "unchecked" }) class SerdeResolverUtilsTests { @Nested class ResolveForType { - private ApplicationContextRunner contextRunner = new ApplicationContextRunner() + private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() .withPropertyValues("spring.cloud.function.ineligible-definitions: sendToDlqAndContinue"); - private Serde fallback = mock(Serde.class); + private final Serde fallback = mock(Serde.class); @Test void returnsFallbackSerdeForWildcard() { @@ -368,7 +368,7 @@ class SerdeResolverUtilsTests { } static class GenericEventSerde implements Serde> { - private String name; + private final String name; GenericEventSerde(String name) { this.name = name; 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 923aec9d6..4462d4bb5 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-2023 the original author or authors. + * Copyright 2018-2024 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. @@ -55,10 +55,11 @@ import static org.assertj.core.api.AssertionsForClassTypes.assertThat; @EmbeddedKafka class KafkaStreamsBinderBootstrapTest { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; @BeforeEach public void before() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); } 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 5e0140c36..9f2447f19 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-2023 the original author or authors. + * Copyright 2021-2024 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. @@ -41,12 +41,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka class KafkaStreamsBinderJaasInitTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static String JAVA_LOGIN_CONFIG_PARAM_VALUE; @BeforeAll public static void beforeAll() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); JAVA_LOGIN_CONFIG_PARAM_VALUE = System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); } 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 93ebb818c..e18186018 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-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -55,12 +55,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"counts", "foo", "bar"}) class KafkaStreamsBinderWordCountBranchesFunctionTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; @BeforeAll public static void setUp() throws Exception { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("groupx", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 a9e1d3977..19dfe654a 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-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -76,7 +76,7 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"counts", "counts-1", "counts-2", "counts-5", "counts-6"}) class KafkaStreamsBinderWordCountFunctionTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; @@ -84,6 +84,7 @@ class KafkaStreamsBinderWordCountFunctionTests { @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 3ebd883b3..477d91005 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-2023 the original author or authors. + * Copyright 2019-2024 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,6 +28,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.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; @@ -47,7 +48,12 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka class KafkaStreamsFunctionStateStoreTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void kafkaStreamsFuncionWithMultipleStateStores() throws Exception { 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 42794bed0..e49b5f46d 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-2023 the original author or authors. + * Copyright 2020-2024 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,6 +26,7 @@ import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.processor.api.Record; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.NoSuchBeanDefinitionException; @@ -55,11 +56,17 @@ import static org.assertj.core.api.AssertionsForInterfaceTypes.assertThat; @EmbeddedKafka class KafkaStreamsRetryTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private final static CountDownLatch LATCH1 = new CountDownLatch(2); + private final static CountDownLatch LATCH2 = new CountDownLatch(4); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } + @Test void retryTemplatePerBindingOnKStream() throws Exception { SpringApplication app = new SpringApplication(RetryTemplatePerConsumerBindingApp.class); 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 b93347ccf..095f6d213 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-2023 the original author or authors. + * Copyright 2019-2024 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,6 +25,7 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serializer; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.KStream; +import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; @@ -55,7 +56,12 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = { "topic1", "topic2" }) class SerdesProvidedAsBeansTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void simpleSerdeBeansAreResolvedProperly() throws Exception { 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 0cf1265b7..b96a53954 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-2023 the original author or authors. + * Copyright 2019-2024 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,6 +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.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; @@ -65,10 +66,15 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "enriched-order") class StreamToGlobalKTableFunctionTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } + @Test void streamToGlobalKTable() throws Exception { SpringApplication app = new SpringApplication(OrderEnricherApplication.class); 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 f3f9f9021..72449280a 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-2023 the original author or authors. + * Copyright 2019-2024 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,6 +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.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; @@ -75,7 +76,12 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "output-topic-1") class StreamToTableJoinFunctionTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void streamToTable() { 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 654ee641b..a83a71459 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-2023 the original author or authors. + * Copyright 2020-2024 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,6 +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.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; @@ -56,7 +57,12 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"topic1-dlq", "topic2-dlq"}) class DlqDestinationResolverTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void dlqDestinationResolverWorks() throws Exception { diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java index 4229c3c7f..bda7f1ba3 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2023-2023 the original author or authors. + * Copyright 2023-2024 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. @@ -52,12 +52,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "hello-dlt-1") class DltAwareProcessorTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); - private static Consumer consumer; + private static EmbeddedKafkaBroker embeddedKafka; + @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); 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 63ed15071..0eb3b4535 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-2023 the original author or authors. + * Copyright 2019-2024 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,12 +46,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"in.1", "in.2", "out"}) class KafkaStreamsBinderDestinationIsPatternTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static org.apache.kafka.clients.consumer.Consumer consumer; @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group", "true", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 5d35fa81e..03427195b 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-2023 the original author or authors. + * Copyright 2019-2024 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. @@ -63,10 +63,11 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"out", "out2"}) class KafkaStreamsBinderHealthIndicatorTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); System.setProperty("logging.level.org.apache.kafka", "OFF"); } 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 c4c70c8e0..8e14655ee 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-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -55,7 +55,7 @@ import static org.assertj.core.api.Assertions.assertThat; * multiple kafka topics(destinations). * * See - * {@link KafkaStreamsBinderMultipleInputTopicsTest#testKstreamWordCountWithStringInputAndPojoOuput} + * {@link KafkaStreamsBinderMultipleInputTopicsTest#kstreamWordCountWithStringInputAndPojoOuput()} * where the input topic names are specified as comma-separated String values for the * property spring.cloud.stream.bindings.input.destination. * @@ -64,12 +64,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "counts") class KafkaStreamsBinderMultipleInputTopicsTest { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; @BeforeAll public static void setUp() throws Exception { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 6860cdab9..87a1230e8 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-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -56,12 +56,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "counts-id") class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; @BeforeAll public static void setUp() throws Exception { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 f773dae99..e4f436ccb 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-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -65,12 +65,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "counts-1") class KafkaStreamsBinderTombstoneTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; @BeforeAll public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 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 c2ee15456..55202eb38 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-2023 the original author or authors. + * Copyright 2018-2024 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,6 +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.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.springframework.boot.SpringApplication; @@ -52,7 +53,12 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka class KafkaStreamsStateStoreIntegrationTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; + + @BeforeAll + public static void setUp() { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + } @Test void testKstreamStateStore() throws Exception { 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 fe74ff206..57d5fdc09 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-2023 the original author or authors. + * Copyright 2017-2024 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. @@ -59,12 +59,13 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = "counts-id") class KafkastreamsBinderPojoInputStringOutputIntegrationTests { - private static final EmbeddedKafkaBroker embeddedKafka = EmbeddedKafkaCondition.getBroker(); + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; @BeforeAll public static void setUp() throws Exception { + embeddedKafka = EmbeddedKafkaCondition.getBroker(); Map consumerProps = KafkaTestUtils.consumerProps("group-id", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");