SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2942: EmbeddedKafka usage improvements in Kafka Streams binder

Browse Source 
* 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
This commit is contained in:
Soby Chacko
2024-05-01 11:52:44 -04:00
parent 1d241583c3
commit d534e840dc
26 changed files with 136 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/EventTypeRoutingWithInferredSerdeTests.java
View File

@@ -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<Integer, Integer> consumer;
private static EmbeddedKafkaBroker embeddedKafka;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

11
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/InteractiveQueryServiceMultiStateStoreTests.java
View File

@@ -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() {

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderEnvironmentPostProcessorTests.java
View File

@@ -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() {

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsEventTypeRoutingTests.java
View File

@@ -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<Integer, Foo> 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<String, Object> consumerProps = KafkaTestUtils.consumerProps("test-group-1", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsFunctionCompositionTests.java
View File

@@ -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<String, String> 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<String, Object> consumerProps = KafkaTestUtils.consumerProps("fn-composition-group", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsInteractiveQueryIntegrationTests.java
View File

@@ -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<String, String> consumer;
private static EmbeddedKafkaBroker embeddedKafka;
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id",
"false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

9
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/MultipleFunctionsInSameAppTests.java
View File

@@ -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<String, String> 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<String, Object> consumerProps = KafkaTestUtils.consumerProps("purchase-groups", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

8
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/SerdeResolverUtilsTests.java
View File

@@ -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<T> implements Serde<GenericEvent<? extends T>> {
private String name;
private final String name;
GenericEventSerde(String name) {
this.name = name;

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java
View File

@@ -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);
}

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderJaasInitTests.java
View File

@@ -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);
}

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java
View File

@@ -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<String, String> consumer;
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("groupx", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java
View File

@@ -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<String, String> consumer;
@@ -84,6 +84,7 @@ class KafkaStreamsBinderWordCountFunctionTests {
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionStateStoreTests.java
View File

@@ -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 {

13
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsRetryTests.java
View File

@@ -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);

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/SerdesProvidedAsBeansTests.java
View File

@@ -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 {

12
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java
View File

@@ -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<Long, EnrichedOrder> consumer;
private static EmbeddedKafkaBroker embeddedKafka;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
}
@Test
void streamToGlobalKTable() throws Exception {
SpringApplication app = new SpringApplication(OrderEnricherApplication.class);

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java
View File

@@ -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() {

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DlqDestinationResolverTests.java
View File

@@ -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 {

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/DltAwareProcessorTests.java
View File

@@ -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<String, String> consumer;
private static EmbeddedKafkaBroker embeddedKafka;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderDestinationIsPatternTests.java
View File

@@ -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<Integer, String> consumer;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "true",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderHealthIndicatorTests.java
View File

@@ -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");
}

7
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderMultipleInputTopicsTest.java
View File

@@ -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<String, String> consumer;
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests.java
View File

@@ -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<Integer, Long> consumer;
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id",
"false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsBinderTombstoneTests.java
View File

@@ -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<String, String> consumer;
@BeforeAll
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group", "false",
embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

10
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsStateStoreIntegrationTests.java
View File

@@ -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 {

5
binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkastreamsBinderPojoInputStringOutputIntegrationTests.java
View File

@@ -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<String, String> consumer;
@BeforeAll
public static void setUp() throws Exception {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group-id",
"false", embeddedKafka);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");