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 6ad1bf559..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 @@ -31,7 +31,6 @@ 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.Disabled; import org.junit.jupiter.api.Test; import org.mockito.Mockito; import org.slf4j.Logger; @@ -62,7 +61,6 @@ import static org.mockito.Mockito.when; * @author Soby Chacko */ @EmbeddedKafka(topics = {"input1", "input2"}) -@Disabled class InteractiveQueryServiceMultiStateStoreTests { private static final String STORE_1_NAME = "store1"; @@ -202,12 +200,12 @@ class InteractiveQueryServiceMultiStateStoreTests { Stores.persistentKeyValueStore(STORE_1_NAME), Serdes.String(), Serdes.String()); } -// @Bean -// public Consumer> app1() { -// return s -> s -// .transformValues(EchoTransformer::new, STORE_1_NAME) -// .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); -// } + @Bean + public Consumer> app1() { + return s -> s + .transformValues(EchoTransformer::new, STORE_1_NAME) + .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); + } @Bean public StoreBuilder> store2() { @@ -215,12 +213,12 @@ class InteractiveQueryServiceMultiStateStoreTests { Stores.persistentKeyValueStore(STORE_2_NAME), Serdes.String(), Serdes.String()); } -// @Bean -// public Consumer> app2() { -// return s -> s -// .transformValues(EchoTransformer::new, STORE_2_NAME) -// .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_2_NAME)); -// } + @Bean + public Consumer> app2() { + return s -> s + .transformValues(EchoTransformer::new, STORE_2_NAME) + .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_2_NAME)); + } @Bean public CleanupConfig cleanupConfig() { @@ -285,12 +283,12 @@ class InteractiveQueryServiceMultiStateStoreTests { Stores.persistentKeyValueStore(STORE_1_NAME), Serdes.String(), Serdes.String()); } -// @Bean -// public Consumer> app1() { -// return s -> s -// .transformValues(EchoTransformer::new, STORE_1_NAME) -// .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); -// } + @Bean + public Consumer> app1() { + return s -> s + .transformValues(EchoTransformer::new, STORE_1_NAME) + .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); + } @Bean public CleanupConfig cleanupConfig() { 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 6fc8ed265..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 @@ -218,9 +218,9 @@ class MultipleFunctionsInSameAppTests { @Bean public Function, KStream[]> processItem() { - return input -> input.split().branch( - (s, p) -> p.equalsIgnoreCase("coffee"); -// (s, p) -> p.equalsIgnoreCase("electronics")); + return input -> input.branch( + (s, p) -> p.equalsIgnoreCase("coffee"), + (s, p) -> p.equalsIgnoreCase("electronics")); } // Testing for the scenario under https://github.com/spring-cloud/spring-cloud-stream/issues/2817 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 808ea3de0..80f2e944b 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 @@ -193,7 +193,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests { final Map> stringKStreamMap = input .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+"))) .groupBy((key, value) -> value) - .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))) + .windowedBy(TimeWindows.of(Duration.ofSeconds(5))) .count(Materialized.as("WordCounts-branch")) .toStream() .map((key, value) -> new KeyValue<>(null, new WordCount(key.key(), value, 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 436190d28..57f0646b7 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 @@ -401,7 +401,7 @@ class KafkaStreamsBinderWordCountFunctionTests { .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) - .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000))) + .windowedBy(TimeWindows.of(Duration.ofMillis(5000))) .count(Materialized.as("foo-WordCounts")) .toStream() .map((key, value) -> new KeyValue<>(key.key(), new WordCount(key.key(), value, diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingDisabledTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingDisabledTests.java index c5b975d13..b67dab139 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingDisabledTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingDisabledTests.java @@ -36,6 +36,7 @@ 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.runner.RunWith; @@ -47,7 +48,7 @@ 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.context.EmbeddedKafka; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; @@ -65,20 +66,17 @@ import static org.mockito.Mockito.verify; @RunWith(SpringRunner.class) @ContextConfiguration @DirtiesContext -@EmbeddedKafka(topics = {"decode-counts", "decode-counts-1"}) public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests { -// /** -// * Kafka rule. -// */ -// @ClassRule -// public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, -// "decode-counts", "decode-counts-1"); -// -// private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule -// .getEmbeddedKafka(); + /** + * Kafka rule. + */ + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "decode-counts", "decode-counts-1"); - private static EmbeddedKafkaBroker embeddedKafka; + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule + .getEmbeddedKafka(); @SpyBean org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingEnabledTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingEnabledTests.java index 7cdc5d37b..5954bf270 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingEnabledTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/KafkaStreamsNativeEncodingDecodingEnabledTests.java @@ -33,6 +33,7 @@ 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.runner.RunWith; @@ -44,7 +45,7 @@ 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.context.EmbeddedKafka; +import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -61,19 +62,17 @@ import static org.mockito.Mockito.verify; @RunWith(SpringRunner.class) @ContextConfiguration @DirtiesContext -@EmbeddedKafka(topics = {"decode-counts", "decode-counts-1"}) public abstract class KafkaStreamsNativeEncodingDecodingEnabledTests { -// /** -// * Kafka rule. -// */ -// @ClassRule -// public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, -// "decode-counts", "decode-counts-1"); + /** + * Kafka rule. + */ + @ClassRule + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, + "decode-counts", "decode-counts-1"); - private static EmbeddedKafkaBroker embeddedKafka; -// = embeddedKafkaRule -// .getEmbeddedKafka(); + private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule + .getEmbeddedKafka(); @SpyBean org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate; 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 e5354152a..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 @@ -23,8 +23,8 @@ import java.util.function.Consumer; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.processor.Processor; import org.apache.kafka.streams.processor.ProcessorContext; -import org.apache.kafka.streams.processor.api.Processor; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.streams.state.WindowStore; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java index 874cf29cc..2828bbaea 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetricsTest.java @@ -163,9 +163,9 @@ class KafkaBinderMetricsTest { MeterFilter.denyNameStartsWith("spring.cloud.stream.binder.kafka.offset")); // Because we have NoopGauge for the offset metric in the meter registry, none of these expectations matter. -// org.mockito.BDDMockito -// .given(consumer.committed(ArgumentMatchers.anySet(TopicPartition.class))) -// .willReturn(new OffsetAndMetadata(500)); + org.mockito.BDDMockito + .given(consumer.committed(ArgumentMatchers.any(TopicPartition.class))) + .willReturn(new OffsetAndMetadata(500)); List partitions = partitions(new Node(0, null, 0)); topicsInUse.put( TEST_TOPIC, diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index cc94b4b91..5157920fc 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -72,7 +72,6 @@ import org.assertj.core.api.Condition; import org.awaitility.Awaitility; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInfo; import org.mockito.ArgumentMatchers; @@ -711,7 +710,6 @@ class KafkaBinderTests extends } @Test - @Disabled @SuppressWarnings({ "unchecked", "rawtypes" }) void testSendAndReceiveBatch() throws Exception { Binder binder = getBinder(); @@ -1628,7 +1626,6 @@ class KafkaBinderTests extends } @Test - @Disabled @SuppressWarnings("unchecked") void configurableDlqName() throws Exception { Binder binder = getBinder(); @@ -2275,7 +2272,6 @@ class KafkaBinderTests extends @Test @Override - @Disabled @SuppressWarnings("unchecked") public void testAnonymousGroup(TestInfo testInfo) throws Exception { Binder binder = getBinder(); @@ -2772,7 +2768,6 @@ class KafkaBinderTests extends } @Test - @Disabled @SuppressWarnings("unchecked") void customPartitionCountOverridesPartitioningIfLarger() throws Exception { var testPayload = new byte[2048]; @@ -3558,7 +3553,7 @@ class KafkaBinderTests extends Consumer consumer = cf.createConsumer(); consumer.subscribe(Collections.singletonList("mixed.0")); - ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); + ConsumerRecords records = consumer.poll(10_1000); Iterator iterator = records.iterator(); ConsumerRecord record = iterator.next(); byte[] value = (byte[]) record.value();