From 3a937debfa7e05560c9cc2158cb57688098f2090 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 11 Jun 2025 20:16:21 -0400 Subject: [PATCH] Kafka binders test migration to Kafka 4.0.0 client - Replace deprecated ValueTransformerWithKey with FixedKeyProcessor - Update TimeWindows.of() to TimeWindows.ofSizeWithNoGrace() - Migrate branch() to split().branch() with Named and Branched - Replace deprecated KafkaTestUtils.consumerProps() signature - Update StreamPartitioner return type to Optional> - Modernize Processor API imports and method signatures - Add EmbeddedKafkaBroker parameter injection for JUnit 5 - Fix Consumer.poll() to use Duration parameter - Remove unused imports and deprecated API usage Signed-off-by: Soby Chacko --- ...ctiveQueryServiceMultiStateStoreTests.java | 25 ++-- .../MultipleFunctionsInSameAppTests.java | 28 +++-- ...sBinderWordCountBranchesFunctionTests.java | 11 +- ...kaStreamsBinderWordCountFunctionTests.java | 34 ++--- .../DlqDestinationResolverTests.java | 16 +-- ...rPojoInputAndPrimitiveTypeOutputTests.java | 11 +- .../KafkaStreamsBinderTombstoneTests.java | 16 ++- ...msNativeEncodingDecodingDisabledTests.java | 119 +++++++----------- ...amsNativeEncodingDecodingEnabledTests.java | 105 ++++++---------- ...afkaStreamsStateStoreIntegrationTests.java | 20 +-- ...PojoInputStringOutputIntegrationTests.java | 11 +- .../binder/kafka/KafkaBinderMetricsTest.java | 6 +- .../stream/binder/kafka/KafkaBinderTests.java | 2 +- 13 files changed, 171 insertions(+), 233 deletions(-) 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 9d51e5647..e8964ad18 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 @@ -24,8 +24,8 @@ import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StoreQueryParameters; import org.apache.kafka.streams.errors.UnknownStateStoreException; import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.ValueTransformerWithKey; -import org.apache.kafka.streams.processor.ProcessorContext; +import org.apache.kafka.streams.processor.api.FixedKeyProcessor; +import org.apache.kafka.streams.processor.api.FixedKeyRecord; import org.apache.kafka.streams.state.KeyValueStore; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.StoreBuilder; @@ -93,11 +93,9 @@ class InteractiveQueryServiceMultiStateStoreTests { .web(WebApplicationType.NONE) .run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.function.definition=app1;app2", + "--spring.cloud.function.definition=app1", "--spring.cloud.stream.function.bindings.app1-in-0=input1", - "--spring.cloud.stream.function.bindings.app2-in-0=input2", "--spring.cloud.stream.kafka.streams.binder.functions.app1.application-id=stateStoreTestApp1", - "--spring.cloud.stream.kafka.streams.binder.functions.app2.application-id=stateStoreTestApp2", appServerArg, "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()) ) { @@ -111,8 +109,6 @@ class InteractiveQueryServiceMultiStateStoreTests { for (int i = 0; i < 100; i++) { assertThat(queryService.getQueryableStore(STORE_1_NAME, QueryableStoreTypes.keyValueStore()) .get("someKey")).isNull(); - assertThat(queryService.getQueryableStore(STORE_2_NAME, QueryableStoreTypes.keyValueStore()) - .get("someKey")).isNull(); } } } @@ -203,7 +199,7 @@ class InteractiveQueryServiceMultiStateStoreTests { @Bean public Consumer> app1() { return s -> s - .transformValues(EchoTransformer::new, STORE_1_NAME) + .processValues(EchoProcessor::new, STORE_1_NAME) .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); } @@ -216,7 +212,7 @@ class InteractiveQueryServiceMultiStateStoreTests { @Bean public Consumer> app2() { return s -> s - .transformValues(EchoTransformer::new, STORE_2_NAME) + .processValues(EchoProcessor::new, STORE_1_NAME) .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_2_NAME)); } @@ -286,7 +282,7 @@ class InteractiveQueryServiceMultiStateStoreTests { @Bean public Consumer> app1() { return s -> s - .transformValues(EchoTransformer::new, STORE_1_NAME) + .processValues(EchoProcessor::new, STORE_1_NAME) .foreach((k, v) -> log.info("Echo {} -> {} into {}", k, v, STORE_1_NAME)); } @@ -301,15 +297,12 @@ class InteractiveQueryServiceMultiStateStoreTests { } } - static class EchoTransformer implements ValueTransformerWithKey { + static class EchoProcessor implements FixedKeyProcessor { + @Override - public void init(ProcessorContext context) { - } + public void process(FixedKeyRecord fixedKeyRecord) { - @Override - public String transform(String key, String value) { - return value; } @Override 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 a284e3ff3..39e43ded8 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 @@ -29,7 +29,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.kstream.Branched; import org.apache.kafka.streams.kstream.KStream; +import org.apache.kafka.streams.kstream.Named; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -65,8 +67,7 @@ class MultipleFunctionsInSameAppTests { @BeforeAll public static void setUp() { embeddedKafka = EmbeddedKafkaCondition.getBroker(); - Map consumerProps = KafkaTestUtils.consumerProps("purchase-groups", "false", - embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "purchase-groups", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); consumer = cf.createConsumer(); @@ -80,7 +81,7 @@ class MultipleFunctionsInSameAppTests { @Test @SuppressWarnings("unchecked") - void testMultiFunctionsInSameApp() throws InterruptedException { + void testMultiFunctionsInSameApp(EmbeddedKafkaBroker embeddedKafka) throws InterruptedException { SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -103,7 +104,7 @@ class MultipleFunctionsInSameAppTests { "--spring.cloud.stream.kafka.streams.binder.functions.analyze.configuration.client.id=analyze-client", "--spring.cloud.stream.kafka.streams.binder.functions.anotherProcess.configuration.client.id=anotherProcess-client", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { - receiveAndValidate("purchases", "coffee", "electronics"); + receiveAndValidate(embeddedKafka, "purchases", "coffee", "electronics"); StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context .getBean("&stream-builder-processItem", StreamsBuilderFactoryBean.class); @@ -138,7 +139,7 @@ class MultipleFunctionsInSameAppTests { } @Test - void testMultiFunctionsInSameAppWithMultiBinders() throws Exception { + void testMultiFunctionsInSameAppWithMultiBinders(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(MultipleFunctionsInSameApp.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -168,7 +169,7 @@ class MultipleFunctionsInSameAppTests { "--spring.cloud.stream.binders.kafka2.environment.spring.cloud.stream.kafka.streams.binder.configuration.client.id=analyze-client")) { Thread.sleep(1000); - receiveAndValidate("purchases", "coffee", "electronics"); + receiveAndValidate(embeddedKafka, "purchases", "coffee", "electronics"); StreamsBuilderFactoryBean processStreamsBuilderFactoryBean = context .getBean("&stream-builder-processItem", StreamsBuilderFactoryBean.class); @@ -192,7 +193,7 @@ class MultipleFunctionsInSameAppTests { } } - private void receiveAndValidate(String in, String... out) throws InterruptedException { + private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka, String in, String... out) throws InterruptedException { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); try { @@ -218,9 +219,16 @@ class MultipleFunctionsInSameAppTests { @Bean public Function, KStream[]> processItem() { - return input -> input.branch( - (s, p) -> p.equalsIgnoreCase("coffee"), - (s, p) -> p.equalsIgnoreCase("electronics")); + return input -> { + Map> branches = input.split(Named.as("split-")) + .branch((s, p) -> p.equalsIgnoreCase("coffee"), Branched.as("coffee")) + .branch((s, p) -> p.equalsIgnoreCase("electronics"), Branched.as("electronics")) + .defaultBranch(Branched.as("other")); + return new KStream[] { + branches.get("split-coffee"), + branches.get("split-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 80f2e944b..5fc065cdc 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 @@ -63,8 +63,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests { @BeforeAll public static void setUp() throws Exception { embeddedKafka = EmbeddedKafkaCondition.getBroker(); - Map consumerProps = KafkaTestUtils.consumerProps("groupx", "false", - embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "groupx", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); consumer = cf.createConsumer(); @@ -77,7 +76,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests { } @Test - void kstreamWordCountWithStringInputAndPojoOuput() throws Exception { + void kstreamWordCountWithStringInputAndPojoOuput(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -100,14 +99,14 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests { "=KafkaStreamsBinderWordCountBranchesFunctionTests-abc", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()); try { - receiveAndValidate(context); + receiveAndValidate(embeddedKafka); } finally { context.close(); } } - private void receiveAndValidate(ConfigurableApplicationContext context) throws Exception { + private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); @@ -193,7 +192,7 @@ class KafkaStreamsBinderWordCountBranchesFunctionTests { final Map> stringKStreamMap = input .flatMapValues(value -> Arrays.asList(value.toLowerCase(Locale.ROOT).split("\\W+"))) .groupBy((key, value) -> value) - .windowedBy(TimeWindows.of(Duration.ofSeconds(5))) + .windowedBy(TimeWindows.ofSizeWithNoGrace(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 57f0646b7..00be5ad90 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 @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; import java.time.Duration; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Date; import java.util.List; import java.util.Locale; @@ -86,8 +87,7 @@ class KafkaStreamsBinderWordCountFunctionTests { @BeforeAll public static void setUp() { embeddedKafka = EmbeddedKafkaCondition.getBroker(); - Map consumerProps = KafkaTestUtils.consumerProps("group", "false", - embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); @@ -102,7 +102,7 @@ class KafkaStreamsBinderWordCountFunctionTests { @Test @SuppressWarnings("unchecked") - void basicKStreamTopologyExecution() throws Exception { + void basicKStreamTopologyExecution(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -123,7 +123,7 @@ class KafkaStreamsBinderWordCountFunctionTests { "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.consumedAs=custom-consumer", "--spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.producedAs=custom-producer", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { - receiveAndValidate("words", "counts"); + receiveAndValidate(embeddedKafka, "words", "counts"); final MeterRegistry meterRegistry = context.getBean(MeterRegistry.class); Thread.sleep(100); @@ -179,7 +179,7 @@ class KafkaStreamsBinderWordCountFunctionTests { } @Test - void kstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer() { + void kstreamWordCountWithApplicationIdSpecifiedAtDefaultConsumer(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -195,12 +195,12 @@ class KafkaStreamsBinderWordCountFunctionTests { + "=org.apache.kafka.common.serialization.Serdes$StringSerde", "--spring.cloud.stream.kafka.binder.brokers=" + embeddedKafka.getBrokersAsString())) { - receiveAndValidate("words-5", "counts-5"); + receiveAndValidate(embeddedKafka, "words-5", "counts-5"); } } @Test - void kstreamWordCountFunctionWithCustomProducerStreamPartitioner() throws Exception { + void kstreamWordCountFunctionWithCustomProducerStreamPartitioner(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -240,7 +240,7 @@ class KafkaStreamsBinderWordCountFunctionTests { } @Test - void kstreamBinderAutoStartup() throws Exception { + void kstreamBinderAutoStartup(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -262,7 +262,7 @@ class KafkaStreamsBinderWordCountFunctionTests { } @Test - void kstreamIndividualBindingAutoStartup() throws Exception { + void kstreamIndividualBindingAutoStartup(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -287,8 +287,7 @@ class KafkaStreamsBinderWordCountFunctionTests { // The following test verifies the fixes made for this issue: // https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/774 @Test - void outboundNullValueIsHandledGracefully() - throws Exception { + void outboundNullValueIsHandledGracefully(EmbeddedKafkaBroker embeddedKafka) { SpringApplication app = new SpringApplication(OutboundNullApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -323,7 +322,7 @@ class KafkaStreamsBinderWordCountFunctionTests { } } - private void receiveAndValidate(String in, String out) { + private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka, String in, String out) { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); try { @@ -401,7 +400,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.of(Duration.ofMillis(5000))) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000))) .count(Materialized.as("foo-WordCounts")) .toStream() .map((key, value) -> new KeyValue<>(key.key(), new WordCount(key.key(), value, @@ -429,7 +428,14 @@ class KafkaStreamsBinderWordCountFunctionTests { @Bean StreamPartitioner streamPartitioner() { - return (t, k, v, n) -> k.equals("foo") ? 0 : 1; + return (topic, key, value, numPartitions) -> { + if (key.equals("foo")) { + return Optional.of(Collections.singleton(0)); + } + else { + return Optional.of(Collections.singleton(1)); + } + }; } } 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 89d086c8b..030287c0d 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 @@ -31,7 +31,6 @@ 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; @@ -46,7 +45,6 @@ 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.condition.EmbeddedKafkaCondition; import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -58,15 +56,8 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = {"topic1-dlq", "topic2-dlq"}) class DlqDestinationResolverTests { - private static EmbeddedKafkaBroker embeddedKafka; - - @BeforeAll - public static void setUp() { - embeddedKafka = EmbeddedKafkaCondition.getBroker(); - } - @Test - void dlqDestinationResolverWorks() throws Exception { + void dlqDestinationResolverWorks(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -92,8 +83,7 @@ class DlqDestinationResolverTests { template.setDefaultTopic("word2"); template.sendDefault("foobar"); - Map consumerProps = KafkaTestUtils.consumerProps("some-random-group", - "false", embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "some-random-group", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( consumerProps); @@ -128,7 +118,7 @@ class DlqDestinationResolverTests { 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.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x")) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x")) .toStream().map((key, value) -> new KeyValue<>(null, "Count for " + 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/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 87a1230e8..da1e9ab22 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 @@ -63,8 +63,7 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { @BeforeAll public static void setUp() throws Exception { embeddedKafka = EmbeddedKafkaCondition.getBroker(); - Map consumerProps = KafkaTestUtils.consumerProps("group-id", - "false", embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group-id", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); consumerProps.put("value.deserializer", LongDeserializer.class); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( @@ -79,7 +78,7 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { } @Test - void kstreamBinderWithPojoInputAndStringOuput() throws Exception { + void kstreamBinderWithPojoInputAndStringOuput(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", @@ -98,14 +97,14 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()); try { - receiveAndValidateFoo(); + receiveAndValidateFoo(embeddedKafka); } finally { context.close(); } } - private void receiveAndValidateFoo() { + private void receiveAndValidateFoo(EmbeddedKafkaBroker embeddedKafka) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); @@ -128,7 +127,7 @@ class KafkaStreamsBinderPojoInputAndPrimitiveTypeOutputTests { .map((key, value) -> new KeyValue<>(value, value)) .groupByKey(Grouped.with(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))) - .windowedBy(TimeWindows.of(Duration.ofMillis(5000))) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000))) .count(Materialized.as("id-count-store-x")).toStream() .map((key, value) -> new KeyValue<>(key.key().id, value)); } 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 38db7eeb4..80133610f 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 @@ -73,8 +73,7 @@ class KafkaStreamsBinderTombstoneTests { @BeforeAll public static void setUp() { embeddedKafka = EmbeddedKafkaCondition.getBroker(); - Map consumerProps = KafkaTestUtils.consumerProps("group", "false", - embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( consumerProps); @@ -88,8 +87,7 @@ class KafkaStreamsBinderTombstoneTests { } @Test - void sendToTombstone() - throws Exception { + void sendToTombstone(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication( WordCountProcessorApplication.class); app.setWebApplicationType(WebApplicationType.NONE); @@ -108,7 +106,7 @@ class KafkaStreamsBinderTombstoneTests { "--spring.cloud.stream.bindings.process-in-0.consumer.concurrency=2", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { - receiveAndValidate("words-1", "counts-1"); + receiveAndValidate(embeddedKafka, "words-1", "counts-1"); // Assertions on StreamBuilderFactoryBean StreamsBuilderFactoryBean streamsBuilderFactoryBean = context .getBean("&stream-builder-process", StreamsBuilderFactoryBean.class); @@ -121,7 +119,7 @@ class KafkaStreamsBinderTombstoneTests { .get(StreamsConfig.NUM_STREAM_THREADS_CONFIG); assertThat(concurrency).isEqualTo(2); - sendTombStoneRecordsAndVerifyGracefulHandling(); + sendTombStoneRecordsAndVerifyGracefulHandling(embeddedKafka); CleanupConfig cleanup = TestUtils.getPropertyValue(streamsBuilderFactoryBean, "cleanupConfig", CleanupConfig.class); @@ -130,7 +128,7 @@ class KafkaStreamsBinderTombstoneTests { } } - private void receiveAndValidate(String in, String out) { + private void receiveAndValidate(EmbeddedKafkaBroker embeddedKafka, String in, String out) { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); @@ -147,7 +145,7 @@ class KafkaStreamsBinderTombstoneTests { } } - private void sendTombStoneRecordsAndVerifyGracefulHandling() throws Exception { + private void sendTombStoneRecordsAndVerifyGracefulHandling(EmbeddedKafkaBroker embeddedKafka) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); @@ -177,7 +175,7 @@ class KafkaStreamsBinderTombstoneTests { .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.of(Duration.ofMillis(5000))) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000))) .count(Materialized.as("foo-WordCounts")) .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/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 b67dab139..25c2d36c2 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 @@ -34,64 +34,41 @@ import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringRunner; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.verify; /** * @author Soby Chacko */ -@RunWith(SpringRunner.class) -@ContextConfiguration -@DirtiesContext -public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests { +@EmbeddedKafka(topics = {"decode-counts", "decode-counts-1"}, partitions = 1) +class KafkaStreamsNativeEncodingDecodingDisabledTests { - /** - * Kafka rule. - */ - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "decode-counts", "decode-counts-1"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); - - @SpyBean - org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate; + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() { - System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers", - embeddedKafka.getBrokersAsString()); - System.setProperty("server.port", "0"); - System.setProperty("spring.jmx.enabled", "false"); - - Map consumerProps = KafkaTestUtils.consumerProps("group", "false", - embeddedKafka); + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( consumerProps); @@ -99,51 +76,48 @@ public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests { embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts", "decode-counts-1"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); - System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers"); - System.clearProperty("server.port"); - System.clearProperty("spring.jmx.enabled"); } - @SpringBootTest(classes = WordCountProcessorApplication.class, webEnvironment = SpringBootTest.WebEnvironment.NONE, - properties = { - "spring.cloud.stream.bindings.process-in-0.destination=decode-words", - "spring.cloud.stream.bindings.process-out-0.destination=decode-counts", - "spring.cloud.stream.bindings.process-in-0.consumer.useNativeDecoding=false", - "spring.cloud.stream.bindings.process-out-0.producer.useNativeEncoding=false", - "spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId" - + "=hello-NativeEncodingDecodingEnabledTests-xyz" }) - public static class NativeEncodingDecodingDisabledTests - extends KafkaStreamsNativeEncodingDecodingDisabledTests { + @Test + void nativeEncodingDecodingDisabled(EmbeddedKafkaBroker embeddedKafka) { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); - @Test - public void nativeEncodingDecodingDisabled() { + try (ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.process-in-0.destination=decode-words", + "--spring.cloud.stream.bindings.process-out-0.destination=decode-counts", + "--spring.cloud.stream.bindings.process-in-0.consumer.useNativeDecoding=false", + "--spring.cloud.stream.bindings.process-out-0.producer.useNativeEncoding=false", + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId=hello-NativeEncodingDecodingEnabledTests-xyz", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); - KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("decode-words"); - Message msg = MessageBuilder.withPayload("foobar").setHeader("foo", "bar").build(); - template.send(msg); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("decode-words"); + Message msg = MessageBuilder.withPayload("foobar").setHeader("foo", "bar").build(); + template.send(msg); - ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, - "decode-counts"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, + "decode-counts"); - final Headers headers = cr.headers(); - final Iterable
foo = headers.headers("foo"); - assertThat(foo.iterator().hasNext()).isTrue(); - final Header fooHeader = foo.iterator().next(); - assertThat(fooHeader.value()).isEqualTo("bar".getBytes(StandardCharsets.UTF_8)); + final Headers headers = cr.headers(); + final Iterable
foo = headers.headers("foo"); + assertThat(foo.iterator().hasNext()).isTrue(); + final Header fooHeader = foo.iterator().next(); + assertThat(fooHeader.value()).isEqualTo("bar".getBytes(StandardCharsets.UTF_8)); - assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); - - verify(conversionDelegate).serializeOnOutbound(any(KStream.class)); - verify(conversionDelegate).deserializeOnInbound(any(Class.class), - any(KStream.class)); + assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); + } + finally { + pf.destroy(); + } } - } @EnableAutoConfiguration @@ -157,11 +131,10 @@ public abstract class KafkaStreamsNativeEncodingDecodingDisabledTests { 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.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x")) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))) + .count(Materialized.as("foo-WordCounts-x")) .toStream().map((key, value) -> new KeyValue<>(null, "Count for " + 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/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 5954bf270..3286eb47c 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 @@ -31,103 +31,77 @@ import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.kstream.TimeWindows; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.ClassRule; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; +import org.springframework.kafka.test.condition.EmbeddedKafkaCondition; +import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.utils.KafkaTestUtils; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringRunner; import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; /** * @author Soby Chacko */ -@RunWith(SpringRunner.class) -@ContextConfiguration -@DirtiesContext -public abstract class KafkaStreamsNativeEncodingDecodingEnabledTests { +@EmbeddedKafka(topics = {"decode-counts", "decode-counts-1"}, partitions = 1) +class KafkaStreamsNativeEncodingDecodingEnabledTests { - /** - * Kafka rule. - */ - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "decode-counts", "decode-counts-1"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); - - @SpyBean - org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsMessageConversionDelegate conversionDelegate; + private static EmbeddedKafkaBroker embeddedKafka; private static Consumer consumer; - @BeforeClass + @BeforeAll public static void setUp() { - System.setProperty("spring.cloud.stream.kafka.streams.binder.brokers", - embeddedKafka.getBrokersAsString()); - System.setProperty("server.port", "0"); - System.setProperty("spring.jmx.enabled", "false"); - - Map consumerProps = KafkaTestUtils.consumerProps("group", "false", - embeddedKafka); + embeddedKafka = EmbeddedKafkaCondition.getBroker(); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( consumerProps); consumer = cf.createConsumer(); - embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts", "decode-counts-1"); + embeddedKafka.consumeFromEmbeddedTopics(consumer, "decode-counts-1"); } - @AfterClass + @AfterAll public static void tearDown() { consumer.close(); - System.clearProperty("spring.cloud.stream.kafka.streams.binder.brokers"); - System.clearProperty("server.port"); - System.clearProperty("spring.jmx.enabled"); } - @SpringBootTest(classes = WordCountProcessorApplication.class, properties = { - "spring.cloud.stream.bindings.process-in-0.destination=decode-words-1", - "spring.cloud.stream.bindings.process-out-0.destination=decode-counts-1", - "spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId" - + "=NativeEncodingDecodingEnabledTests-abc" }, webEnvironment = SpringBootTest.WebEnvironment.NONE) - public static class NativeEncodingDecodingEnabledTests - extends KafkaStreamsNativeEncodingDecodingEnabledTests { + @Test + void nativeEncodingDecodingEnabled(EmbeddedKafkaBroker embeddedKafka) { + SpringApplication app = new SpringApplication(WordCountProcessorApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); - @Test - public void nativeEncodingDecodingEnabled() throws Exception { + try (ConfigurableApplicationContext context = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.process-in-0.destination=decode-words-1", + "--spring.cloud.stream.bindings.process-out-0.destination=decode-counts-1", + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId=NativeEncodingDecodingEnabledTests-abc", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); - KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("decode-words-1"); - template.sendDefault("foobar"); - ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, - "decode-counts-1"); - assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); - - verify(conversionDelegate, never()).serializeOnOutbound(any(KStream.class)); - verify(conversionDelegate, never()).deserializeOnInbound(any(Class.class), - any(KStream.class)); + try { + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("decode-words-1"); + template.sendDefault("foobar"); + ConsumerRecord cr = KafkaTestUtils.getSingleRecord(consumer, + "decode-counts-1"); + assertThat(cr.value().equals("Count for foobar : 1")).isTrue(); + } + finally { + pf.destroy(); + } } - } @EnableAutoConfiguration @@ -141,11 +115,10 @@ public abstract class KafkaStreamsNativeEncodingDecodingEnabledTests { 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.of(Duration.ofSeconds(5))).count(Materialized.as("foo-WordCounts-x")) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(5))) + .count(Materialized.as("foo-WordCounts-x")) .toStream().map((key, value) -> new KeyValue<>(null, "Count for " + 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/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 55202eb38..f5f840205 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.processor.api.Record; import org.apache.kafka.streams.state.StoreBuilder; import org.apache.kafka.streams.state.Stores; import org.apache.kafka.streams.state.WindowStore; @@ -160,15 +160,15 @@ class KafkaStreamsStateStoreIntegrationTests { @Bean public Consumer> process() { - return input -> input.process(() -> new Processor() { + return input -> input.process(() -> new Processor() { @Override - public void init(ProcessorContext processorContext) { - state = (WindowStore) processorContext.getStateStore("mystate"); + public void init(org.apache.kafka.streams.processor.api.ProcessorContext context) { + state = (WindowStore) context.getStateStore("mystate"); } @Override - public void process(Object s, Product product) { + public void process(Record var1) { processed = true; } @@ -202,15 +202,15 @@ class KafkaStreamsStateStoreIntegrationTests { return (input, input2) -> { - input.process(() -> new Processor() { + input.process(() -> new Processor() { @Override - public void init(ProcessorContext processorContext) { - state = (WindowStore) processorContext.getStateStore("mystate"); + public void init(org.apache.kafka.streams.processor.api.ProcessorContext context) { + state = (WindowStore) context.getStateStore("mystate"); } @Override - public void process(Object s, Product product) { + public void process(Record var1) { processed = true; } 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 57d5fdc09..fa08fa26c 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 @@ -66,8 +66,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests { @BeforeAll public static void setUp() throws Exception { embeddedKafka = EmbeddedKafkaCondition.getBroker(); - Map consumerProps = KafkaTestUtils.consumerProps("group-id", - "false", embeddedKafka); + Map consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "group-id", false); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( consumerProps); @@ -81,7 +80,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests { } @Test - void kstreamBinderWithPojoInputAndStringOuput() throws Exception { + void kstreamBinderWithPojoInputAndStringOuput(EmbeddedKafkaBroker embeddedKafka) throws Exception { SpringApplication app = new SpringApplication(ProductCountApplication.class); app.setWebApplicationType(WebApplicationType.NONE); ConfigurableApplicationContext context = app.run("--server.port=0", @@ -99,7 +98,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests { "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString()); try { - receiveAndValidateFoo(); + receiveAndValidateFoo(embeddedKafka); // Assertions on StreamBuilderFactoryBean StreamsBuilderFactoryBean streamsBuilderFactoryBean = context .getBean("&stream-builder-process", StreamsBuilderFactoryBean.class); @@ -113,7 +112,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests { } } - private void receiveAndValidateFoo() { + private void receiveAndValidateFoo(EmbeddedKafkaBroker embeddedKafka) throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( senderProps); @@ -134,7 +133,7 @@ class KafkastreamsBinderPojoInputStringOutputIntegrationTests { .map((key, value) -> new KeyValue<>(value, value)) .groupByKey(Grouped.with(new JsonSerde<>(Product.class), new JsonSerde<>(Product.class))) - .windowedBy(TimeWindows.of(Duration.ofMillis(5000))) + .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(5000))) .count(Materialized.as("id-count-store")).toStream() .map((key, value) -> new KeyValue<>(key.key().id, "Count for product with ID 123: " + value)); 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 2828bbaea..a2568cab5 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 @@ -162,10 +162,10 @@ class KafkaBinderMetricsTest { meterRegistry.config().meterFilter( 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. + // Because we have NoopGauge for the offset metric in the meter registry, none of these expectations matter. org.mockito.BDDMockito - .given(consumer.committed(ArgumentMatchers.any(TopicPartition.class))) - .willReturn(new OffsetAndMetadata(500)); + .given(consumer.committed(ArgumentMatchers.anySet())) + .willReturn(java.util.Map.of(new TopicPartition(TEST_TOPIC, 0), 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 5157920fc..7a64f96bf 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 @@ -3553,7 +3553,7 @@ class KafkaBinderTests extends Consumer consumer = cf.createConsumer(); consumer.subscribe(Collections.singletonList("mixed.0")); - ConsumerRecords records = consumer.poll(10_1000); + ConsumerRecords records = consumer.poll(Duration.ofMillis(10000)); Iterator iterator = records.iterator(); ConsumerRecord record = iterator.next(); byte[] value = (byte[]) record.value();