From adde49aab3744fe129b40177abd478be8d566071 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 25 Aug 2021 15:45:25 -0400 Subject: [PATCH] Refactoring Kafka Streams join tests Coalesce the stream join tests in Kafka Streams binder around the functional model. Remove duplicating the same join tests using the StreamListener model. --- .../StreamToGlobalKTableFunctionTests.java | 45 +- .../StreamToTableJoinFunctionTests.java | 92 +++- ...eamToGlobalKTableJoinIntegrationTests.java | 383 -------------- .../StreamToTableJoinIntegrationTests.java | 497 ------------------ 4 files changed, 134 insertions(+), 883 deletions(-) delete mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java delete mode 100644 spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java index 84648e36b..b7eb33172 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java @@ -41,7 +41,13 @@ import org.junit.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; +import org.springframework.cloud.stream.binder.ProducerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBindingProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; @@ -70,7 +76,7 @@ public class StreamToGlobalKTableFunctionTests { public void testStreamToGlobalKTable() throws Exception { SpringApplication app = new SpringApplication(OrderEnricherApplication.class); app.setWebApplicationType(WebApplicationType.NONE); - try (ConfigurableApplicationContext ignored = app.run("--server.port=0", + try (ConfigurableApplicationContext context = app.run("--server.port=0", "--spring.jmx.enabled=false", "--spring.cloud.stream.function.definition=process", "--spring.cloud.stream.function.bindings.process-in-0=order", @@ -89,7 +95,44 @@ public class StreamToGlobalKTableFunctionTests { "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", "--spring.cloud.stream.kafka.streams.bindings.order.consumer.applicationId=" + "StreamToGlobalKTableJoinFunctionTests-abc", + + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.topic.properties.cleanup.policy=compact", + "--spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.topic.properties.cleanup.policy=compact", + "--spring.cloud.stream.kafka.streams.bindings.process-in-2.consumer.topic.properties.cleanup.policy=compact", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { + + // Testing certain ancillary configuration of GlobalKTable around topics creation. + // See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/687 + + BinderFactory binderFactory = context.getBeanFactory() + .getBean(BinderFactory.class); + + Binder kStreamBinder = binderFactory + .getBinder("kstream", KStream.class); + + KafkaStreamsConsumerProperties input = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) kStreamBinder) + .getExtendedConsumerProperties("process-in-0"); + String cleanupPolicy = input.getTopic().getProperties().get("cleanup.policy"); + + assertThat(cleanupPolicy).isEqualTo("compact"); + + Binder globalKTableBinder = binderFactory + .getBinder("globalktable", GlobalKTable.class); + + KafkaStreamsConsumerProperties inputX = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) globalKTableBinder) + .getExtendedConsumerProperties("process-in-1"); + String cleanupPolicyX = inputX.getTopic().getProperties().get("cleanup.policy"); + + assertThat(cleanupPolicyX).isEqualTo("compact"); + + KafkaStreamsConsumerProperties inputY = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) globalKTableBinder) + .getExtendedConsumerProperties("process-in-2"); + String cleanupPolicyY = inputY.getTopic().getProperties().get("cleanup.policy"); + + assertThat(cleanupPolicyY).isEqualTo("compact"); + + Map senderPropsCustomer = KafkaTestUtils.producerProps(embeddedKafka); senderPropsCustomer.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class); senderPropsCustomer.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java index b235984d9..59aa576f7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -39,16 +40,25 @@ import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.kstream.Grouped; +import org.apache.kafka.streams.kstream.JoinWindows; import org.apache.kafka.streams.kstream.Joined; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; +import org.apache.kafka.streams.kstream.StreamJoined; import org.junit.ClassRule; import org.junit.Test; import org.springframework.boot.SpringApplication; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; +import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.kafka.core.CleanupConfig; @@ -176,9 +186,8 @@ public class StreamToTableJoinFunctionTests { } } - private void runTest(SpringApplication app, Consumer consumer) { - try (ConfigurableApplicationContext ignored = app.run("--server.port=0", + try (ConfigurableApplicationContext context = app.run("--server.port=0", "--spring.jmx.enabled=false", "--spring.cloud.stream.bindings.process-in-0.destination=user-clicks-1", "--spring.cloud.stream.bindings.process-in-1.destination=user-regions-1", @@ -190,6 +199,8 @@ public class StreamToTableJoinFunctionTests { "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.applicationId" + "=StreamToTableJoinFunctionTests-abc", + "--spring.cloud.stream.kafka.streams.bindings.process-in-1.consumer.topic.properties.cleanup.policy=compact", + "--spring.cloud.stream.kafka.streams.bindings.process-out-0.producer.topic.properties.cleanup.policy=compact", "--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString())) { // Input 1: Region per user (multiple records allowed per user). @@ -259,6 +270,30 @@ public class StreamToTableJoinFunctionTests { assertThat(count == expectedClicksPerRegion.size()).isTrue(); assertThat(actualClicksPerRegion).hasSameElementsAs(expectedClicksPerRegion); + + // Testing certain ancillary configuration of GlobalKTable around topics creation. + // See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/687 + BinderFactory binderFactory = context.getBeanFactory() + .getBean(BinderFactory.class); + + Binder ktableBinder = binderFactory + .getBinder("ktable", KTable.class); + + KafkaStreamsConsumerProperties inputX = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) ktableBinder) + .getExtendedConsumerProperties("process-in-1"); + String cleanupPolicyX = inputX.getTopic().getProperties().get("cleanup.policy"); + + assertThat(cleanupPolicyX).isEqualTo("compact"); + + Binder kStreamBinder = binderFactory + .getBinder("kstream", KStream.class); + + KafkaStreamsProducerProperties producerProperties = (KafkaStreamsProducerProperties) ((ExtendedPropertiesBinder) kStreamBinder) + .getExtendedProducerProperties("process-out-0"); + + String cleanupPolicyOutput = producerProperties.getTopic().getProperties().get("cleanup.policy"); + + assertThat(cleanupPolicyOutput).isEqualTo("compact"); } finally { consumer.close(); @@ -401,6 +436,34 @@ public class StreamToTableJoinFunctionTests { } } + @Test + public void testTrivialSingleKTableInputAsNonDeclarative() { + SpringApplication app = new SpringApplication(TrivialKTableApp.class); + app.setWebApplicationType(WebApplicationType.NONE); + app.run("--server.port=0", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString(), + "--spring.cloud.stream.kafka.streams.bindings.process-in-0.consumer.application-id=" + + "testTrivialSingleKTableInputAsNonDeclarative"); + //All we are verifying is that this application didn't throw any errors. + //See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/536 + } + + @Test + public void testTwoKStreamsCanBeJoined() { + SpringApplication app = new SpringApplication( + JoinProcessor.class); + app.setWebApplicationType(WebApplicationType.NONE); + app.run("--server.port=0", + "--spring.cloud.stream.kafka.streams.binder.brokers=" + + embeddedKafka.getBrokersAsString(), + "--spring.application.name=" + + "two-kstream-input-join-integ-test"); + //All we are verifying is that this application didn't throw any errors. + //See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/701 + } + + /** * Tuple for a region and its associated number of clicks. */ @@ -488,4 +551,29 @@ public class StreamToTableJoinFunctionTests { } } + @EnableAutoConfiguration + public static class TrivialKTableApp { + + public java.util.function.Consumer> process() { + return inputTable -> inputTable.toStream().foreach((key, value) -> System.out.println("key : value " + key + " : " + value)); + } + } + + @EnableAutoConfiguration + public static class JoinProcessor { + + public BiConsumer, KStream> testProcessor() { + return (input1Stream, input2Stream) -> input1Stream + .join(input2Stream, + (event1, event2) -> null, + JoinWindows.of(Duration.ofMillis(5)), + StreamJoined.with( + Serdes.String(), + Serdes.String(), + Serdes.String() + ) + ); + } + } + } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java deleted file mode 100644 index ff0ac78c0..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToGlobalKTableJoinIntegrationTests.java +++ /dev/null @@ -1,383 +0,0 @@ -/* - * Copyright 2018-2019 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. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka.streams.integration; - -import java.util.ArrayList; -import java.util.Comparator; -import java.util.List; -import java.util.Map; - -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.common.serialization.LongDeserializer; -import org.apache.kafka.common.serialization.LongSerializer; -import org.apache.kafka.streams.KeyValue; -import org.apache.kafka.streams.kstream.GlobalKTable; -import org.apache.kafka.streams.kstream.KStream; -import org.junit.ClassRule; -import org.junit.Test; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.BinderFactory; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; -import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.kafka.core.DefaultKafkaConsumerFactory; -import org.springframework.kafka.core.DefaultKafkaProducerFactory; -import org.springframework.kafka.core.KafkaTemplate; -import org.springframework.kafka.support.serializer.JsonDeserializer; -import org.springframework.kafka.support.serializer.JsonSerializer; -import org.springframework.kafka.test.EmbeddedKafkaBroker; -import org.springframework.kafka.test.rule.EmbeddedKafkaRule; -import org.springframework.kafka.test.utils.KafkaTestUtils; -import org.springframework.messaging.handler.annotation.SendTo; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Soby Chacko - */ -public class StreamToGlobalKTableJoinIntegrationTests { - - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "enriched-order"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); - - private static Consumer consumer; - - @Test - public void testStreamToGlobalKTable() throws Exception { - SpringApplication app = new SpringApplication( - StreamToGlobalKTableJoinIntegrationTests.OrderEnricherApplication.class); - app.setWebApplicationType(WebApplicationType.NONE); - ConfigurableApplicationContext context = app.run("--server.port=0", - "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.destination=orders", - "--spring.cloud.stream.bindings.input-x.destination=customers", - "--spring.cloud.stream.bindings.input-y.destination=products", - "--spring.cloud.stream.bindings.output.destination=enriched-order", - "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" - + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" - + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", - "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId" - + "=StreamToGlobalKTableJoinIntegrationTests-abc", - "--spring.cloud.stream.kafka.streams.bindings.input.consumer.topic.properties.cleanup.policy=compact", - "--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.topic.properties.cleanup.policy=compact", - "--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.topic.properties.cleanup.policy=compact", - "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString()); - try { - // Testing certain ancillary configuration of GlobalKTable around topics creation. - // See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/687 - - BinderFactory binderFactory = context.getBeanFactory() - .getBean(BinderFactory.class); - - Binder kStreamBinder = binderFactory - .getBinder("kstream", KStream.class); - - KafkaStreamsConsumerProperties input = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) kStreamBinder) - .getExtendedConsumerProperties("input"); - String cleanupPolicy = input.getTopic().getProperties().get("cleanup.policy"); - - assertThat(cleanupPolicy).isEqualTo("compact"); - - Binder globalKTableBinder = binderFactory - .getBinder("globalktable", GlobalKTable.class); - - KafkaStreamsConsumerProperties inputX = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) globalKTableBinder) - .getExtendedConsumerProperties("input-x"); - String cleanupPolicyX = inputX.getTopic().getProperties().get("cleanup.policy"); - - assertThat(cleanupPolicyX).isEqualTo("compact"); - - KafkaStreamsConsumerProperties inputY = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) globalKTableBinder) - .getExtendedConsumerProperties("input-y"); - String cleanupPolicyY = inputY.getTopic().getProperties().get("cleanup.policy"); - - assertThat(cleanupPolicyY).isEqualTo("compact"); - - Map senderPropsCustomer = KafkaTestUtils - .producerProps(embeddedKafka); - senderPropsCustomer.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - LongSerializer.class); - senderPropsCustomer.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - JsonSerializer.class); - - DefaultKafkaProducerFactory pfCustomer = new DefaultKafkaProducerFactory<>( - senderPropsCustomer); - KafkaTemplate template = new KafkaTemplate<>(pfCustomer, - true); - template.setDefaultTopic("customers"); - for (long i = 0; i < 5; i++) { - final Customer customer = new Customer(); - customer.setName("customer-" + i); - template.sendDefault(i, customer); - } - - Map senderPropsProduct = KafkaTestUtils - .producerProps(embeddedKafka); - senderPropsProduct.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - LongSerializer.class); - senderPropsProduct.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - JsonSerializer.class); - - DefaultKafkaProducerFactory pfProduct = new DefaultKafkaProducerFactory<>( - senderPropsProduct); - KafkaTemplate productTemplate = new KafkaTemplate<>(pfProduct, - true); - productTemplate.setDefaultTopic("products"); - - for (long i = 0; i < 5; i++) { - final Product product = new Product(); - product.setName("product-" + i); - productTemplate.sendDefault(i, product); - } - - Map senderPropsOrder = KafkaTestUtils - .producerProps(embeddedKafka); - senderPropsOrder.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - LongSerializer.class); - senderPropsOrder.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - JsonSerializer.class); - - DefaultKafkaProducerFactory pfOrder = new DefaultKafkaProducerFactory<>( - senderPropsOrder); - KafkaTemplate orderTemplate = new KafkaTemplate<>(pfOrder, true); - orderTemplate.setDefaultTopic("orders"); - - for (long i = 0; i < 5; i++) { - final Order order = new Order(); - order.setCustomerId(i); - order.setProductId(i); - orderTemplate.sendDefault(i, order); - } - - Map consumerProps = KafkaTestUtils.consumerProps("group", - "false", embeddedKafka); - consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, - LongDeserializer.class); - consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, - JsonDeserializer.class); - consumerProps.put(JsonDeserializer.VALUE_DEFAULT_TYPE, - "org.springframework.cloud.stream.binder.kafka.streams.integration." - + "StreamToGlobalKTableJoinIntegrationTests.EnrichedOrder"); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( - consumerProps); - - consumer = cf.createConsumer(); - embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "enriched-order"); - - int count = 0; - long start = System.currentTimeMillis(); - List> enrichedOrders = new ArrayList<>(); - do { - ConsumerRecords records = KafkaTestUtils - .getRecords(consumer); - count = count + records.count(); - for (ConsumerRecord record : records) { - enrichedOrders.add(new KeyValue<>(record.key(), record.value())); - } - } - while (count < 5 && (System.currentTimeMillis() - start) < 30000); - - assertThat(count == 5).isTrue(); - assertThat(enrichedOrders.size() == 5).isTrue(); - - enrichedOrders.sort(Comparator.comparing(o -> o.key)); - - for (int i = 0; i < 5; i++) { - KeyValue enrichedOrderKeyValue = enrichedOrders - .get(i); - assertThat(enrichedOrderKeyValue.key == i).isTrue(); - EnrichedOrder enrichedOrder = enrichedOrderKeyValue.value; - assertThat(enrichedOrder.getOrder().customerId == i).isTrue(); - assertThat(enrichedOrder.getOrder().productId == i).isTrue(); - assertThat(enrichedOrder.getCustomer().name.equals("customer-" + i)) - .isTrue(); - assertThat(enrichedOrder.getProduct().name.equals("product-" + i)) - .isTrue(); - } - pfCustomer.destroy(); - pfProduct.destroy(); - pfOrder.destroy(); - consumer.close(); - } - finally { - context.close(); - } - } - - interface CustomGlobalKTableProcessor extends KafkaStreamsProcessor { - - @Input("input-x") - GlobalKTable inputX(); - - @Input("input-y") - GlobalKTable inputY(); - - } - - @EnableBinding(CustomGlobalKTableProcessor.class) - @EnableAutoConfiguration - public static class OrderEnricherApplication { - - @StreamListener - @SendTo("output") - public KStream process( - @Input("input") KStream ordersStream, - @Input("input-x") GlobalKTable customers, - @Input("input-y") GlobalKTable products) { - - KStream customerOrdersStream = ordersStream.join( - customers, (orderId, order) -> order.getCustomerId(), - (order, customer) -> new CustomerOrder(customer, order)); - - return customerOrdersStream.join(products, - (orderId, customerOrder) -> customerOrder.productId(), - (customerOrder, product) -> { - EnrichedOrder enrichedOrder = new EnrichedOrder(); - enrichedOrder.setProduct(product); - enrichedOrder.setCustomer(customerOrder.customer); - enrichedOrder.setOrder(customerOrder.order); - return enrichedOrder; - }); - } - - } - - static class Order { - - long customerId; - - long productId; - - public long getCustomerId() { - return customerId; - } - - public void setCustomerId(long customerId) { - this.customerId = customerId; - } - - public long getProductId() { - return productId; - } - - public void setProductId(long productId) { - this.productId = productId; - } - - } - - static class Customer { - - String name; - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - } - - static class Product { - - String name; - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - } - - static class EnrichedOrder { - - Product product; - - Customer customer; - - Order order; - - public Product getProduct() { - return product; - } - - public void setProduct(Product product) { - this.product = product; - } - - public Customer getCustomer() { - return customer; - } - - public void setCustomer(Customer customer) { - this.customer = customer; - } - - public Order getOrder() { - return order; - } - - public void setOrder(Order order) { - this.order = order; - } - - } - - private static class CustomerOrder { - - private final Customer customer; - - private final Order order; - - CustomerOrder(final Customer customer, final Order order) { - this.customer = customer; - this.order = order; - } - - long productId() { - return order.getProductId(); - } - - } -} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java deleted file mode 100644 index d2e67ab19..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java +++ /dev/null @@ -1,497 +0,0 @@ -/* - * Copyright 2018-2019 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. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.binder.kafka.streams.integration; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; -import java.util.Map; -import java.util.concurrent.TimeUnit; - -import org.apache.kafka.clients.consumer.Consumer; -import org.apache.kafka.clients.consumer.ConsumerConfig; -import org.apache.kafka.clients.consumer.ConsumerRecord; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.producer.ProducerConfig; -import org.apache.kafka.common.serialization.LongDeserializer; -import org.apache.kafka.common.serialization.LongSerializer; -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.common.serialization.StringDeserializer; -import org.apache.kafka.common.serialization.StringSerializer; -import org.apache.kafka.streams.KeyValue; -import org.apache.kafka.streams.kstream.JoinWindows; -import org.apache.kafka.streams.kstream.Joined; -import org.apache.kafka.streams.kstream.KStream; -import org.apache.kafka.streams.kstream.KTable; -import org.apache.kafka.streams.kstream.Serialized; -import org.junit.ClassRule; -import org.junit.Test; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.StreamListener; -import org.springframework.cloud.stream.binder.Binder; -import org.springframework.cloud.stream.binder.BinderFactory; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsProcessor; -import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; -import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsProducerProperties; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.kafka.core.CleanupConfig; -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.utils.KafkaTestUtils; -import org.springframework.messaging.handler.annotation.SendTo; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Soby Chacko - */ -public class StreamToTableJoinIntegrationTests { - - @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, - "output-topic-1", "output-topic-2", "user-clicks-2", "user-regions-2"); - - private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule - .getEmbeddedKafka(); - - @Test - public void testStreamToTable() throws Exception { - SpringApplication app = new SpringApplication( - CountClicksPerRegionApplication.class); - app.setWebApplicationType(WebApplicationType.NONE); - - Consumer consumer; - Map consumerProps = KafkaTestUtils.consumerProps("group-1", - "false", embeddedKafka); - consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, - StringDeserializer.class); - consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, - LongDeserializer.class); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( - consumerProps); - consumer = cf.createConsumer(); - embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic-1"); - - ConfigurableApplicationContext context = app.run("--server.port=0", - "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.destination=user-clicks-1", - "--spring.cloud.stream.bindings.input-x.destination=user-regions-1", - "--spring.cloud.stream.bindings.output.destination=output-topic-1", - "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" - + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" - + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", - "--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId" - + "=StreamToTableJoinIntegrationTests-abc", - "--spring.cloud.stream.kafka.streams.bindings.input-x.consumer.topic.properties.cleanup.policy=compact", - "--spring.cloud.stream.kafka.streams.bindings.output.producer.topic.properties.cleanup.policy=compact", - "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString()); - try { - // Testing certain ancillary configuration of GlobalKTable around topics creation. - // See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/687 - BinderFactory binderFactory = context.getBeanFactory() - .getBean(BinderFactory.class); - - Binder ktableBinder = binderFactory - .getBinder("ktable", KTable.class); - - KafkaStreamsConsumerProperties inputX = (KafkaStreamsConsumerProperties) ((ExtendedPropertiesBinder) ktableBinder) - .getExtendedConsumerProperties("input-x"); - String cleanupPolicyX = inputX.getTopic().getProperties().get("cleanup.policy"); - - assertThat(cleanupPolicyX).isEqualTo("compact"); - - Binder kStreamBinder = binderFactory - .getBinder("kstream", KStream.class); - - KafkaStreamsProducerProperties producerProperties = (KafkaStreamsProducerProperties) ((ExtendedPropertiesBinder) kStreamBinder) - .getExtendedProducerProperties("output"); - - String cleanupPolicyOutput = producerProperties.getTopic().getProperties().get("cleanup.policy"); - - assertThat(cleanupPolicyOutput).isEqualTo("compact"); - - // Input 1: Region per user (multiple records allowed per user). - List> userRegions = Arrays.asList(new KeyValue<>( - "alice", "asia"), /* Alice lived in Asia originally... */ - new KeyValue<>("bob", "americas"), new KeyValue<>("chao", "asia"), - new KeyValue<>("dave", "europe"), new KeyValue<>("alice", - "europe"), /* ...but moved to Europe some time later. */ - new KeyValue<>("eve", "americas"), new KeyValue<>("fang", "asia")); - - Map senderProps1 = KafkaTestUtils - .producerProps(embeddedKafka); - senderProps1.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - StringSerializer.class); - senderProps1.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - StringSerializer.class); - - DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>( - senderProps1); - KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); - template1.setDefaultTopic("user-regions-1"); - - for (KeyValue keyValue : userRegions) { - template1.sendDefault(keyValue.key, keyValue.value); - } - - // Input 2: Clicks per user (multiple records allowed per user). - List> userClicks = Arrays.asList( - new KeyValue<>("alice", 13L), new KeyValue<>("bob", 4L), - new KeyValue<>("chao", 25L), new KeyValue<>("bob", 19L), - new KeyValue<>("dave", 56L), new KeyValue<>("eve", 78L), - new KeyValue<>("alice", 40L), new KeyValue<>("fang", 99L)); - - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - StringSerializer.class); - senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - LongSerializer.class); - - DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( - senderProps); - KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("user-clicks-1"); - - for (KeyValue keyValue : userClicks) { - template.sendDefault(keyValue.key, keyValue.value); - } - - List> expectedClicksPerRegion = Arrays.asList( - new KeyValue<>("americas", 101L), new KeyValue<>("europe", 109L), - new KeyValue<>("asia", 124L)); - - // Verify that we receive the expected data - int count = 0; - long start = System.currentTimeMillis(); - List> actualClicksPerRegion = new ArrayList<>(); - do { - ConsumerRecords records = KafkaTestUtils - .getRecords(consumer); - count = count + records.count(); - for (ConsumerRecord record : records) { - actualClicksPerRegion - .add(new KeyValue<>(record.key(), record.value())); - } - } - while (count < expectedClicksPerRegion.size() - && (System.currentTimeMillis() - start) < 30000); - - assertThat(count == expectedClicksPerRegion.size()).isTrue(); - assertThat(actualClicksPerRegion).hasSameElementsAs(expectedClicksPerRegion); - } - finally { - consumer.close(); - } - } - - @Test - public void testGlobalStartOffsetWithLatestAndIndividualBindingWthEarliest() - throws Exception { - SpringApplication app = new SpringApplication( - CountClicksPerRegionApplication.class); - app.setWebApplicationType(WebApplicationType.NONE); - - Consumer consumer; - Map consumerProps = KafkaTestUtils.consumerProps("group-2", - "false", embeddedKafka); - consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, - StringDeserializer.class); - consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, - LongDeserializer.class); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>( - consumerProps); - consumer = cf.createConsumer(); - embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic-2"); - - // Produce data first to the input topic to test the startOffset setting on the - // binding (which is set to earliest below). - // Input 1: Clicks per user (multiple records allowed per user). - List> userClicks = Arrays.asList( - new KeyValue<>("alice", 100L), new KeyValue<>("alice", 100L), - new KeyValue<>("alice", 100L), new KeyValue<>("alice", 100L), - new KeyValue<>("alice", 100L), new KeyValue<>("alice", 100L), - new KeyValue<>("alice", 100L), new KeyValue<>("alice", 100L), - new KeyValue<>("alice", 100L), new KeyValue<>("alice", 100L)); - - Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); - senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - StringSerializer.class); - senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - LongSerializer.class); - - DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>( - senderProps); - KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("user-clicks-2"); - - for (KeyValue keyValue : userClicks) { - template.sendDefault(keyValue.key, keyValue.value); - } - // Thread.sleep(10000L); - try (ConfigurableApplicationContext ignored = app.run("--server.port=0", - "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.destination=user-clicks-2", - "--spring.cloud.stream.bindings.input-x.destination=user-regions-2", - "--spring.cloud.stream.bindings.output.destination=output-topic-2", - "--spring.cloud.stream.kafka.streams.binder.configuration.auto.offset.reset=latest", - "--spring.cloud.stream.kafka.streams.bindings.input.consumer.startOffset=earliest", - "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde" - + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde" - + "=org.apache.kafka.common.serialization.Serdes$StringSerde", - "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", - "--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=helloxyz-foobar", - "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString())) { - Thread.sleep(1000L); - - // Input 2: Region per user (multiple records allowed per user). - List> userRegions = Arrays.asList(new KeyValue<>( - "alice", "asia"), /* Alice lived in Asia originally... */ - new KeyValue<>("bob", "americas"), new KeyValue<>("chao", "asia"), - new KeyValue<>("dave", "europe"), new KeyValue<>("alice", - "europe"), /* ...but moved to Europe some time later. */ - new KeyValue<>("eve", "americas"), new KeyValue<>("fang", "asia")); - - Map senderProps1 = KafkaTestUtils - .producerProps(embeddedKafka); - senderProps1.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, - StringSerializer.class); - senderProps1.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, - StringSerializer.class); - - DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>( - senderProps1); - KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); - template1.setDefaultTopic("user-regions-2"); - - for (KeyValue keyValue : userRegions) { - template1.sendDefault(keyValue.key, keyValue.value); - } - - // Input 1: Clicks per user (multiple records allowed per user). - List> userClicks1 = Arrays.asList( - new KeyValue<>("bob", 4L), new KeyValue<>("chao", 25L), - new KeyValue<>("bob", 19L), new KeyValue<>("dave", 56L), - new KeyValue<>("eve", 78L), new KeyValue<>("fang", 99L)); - - for (KeyValue keyValue : userClicks1) { - template.sendDefault(keyValue.key, keyValue.value); - } - - List> expectedClicksPerRegion = Arrays.asList( - new KeyValue<>("americas", 101L), new KeyValue<>("europe", 56L), - new KeyValue<>("asia", 124L), - // 1000 alice entries which were there in the topic before the - // consumer started. - // Since we set the startOffset to earliest for the topic, it will - // read them, - // but the join fails to associate with a valid region, thus UNKNOWN. - new KeyValue<>("UNKNOWN", 1000L)); - - // Verify that we receive the expected data - int count = 0; - long start = System.currentTimeMillis(); - List> actualClicksPerRegion = new ArrayList<>(); - do { - ConsumerRecords records = KafkaTestUtils - .getRecords(consumer); - count = count + records.count(); - for (ConsumerRecord record : records) { - System.out.println("foobar: " + record.key() + "::" + record.value()); - actualClicksPerRegion - .add(new KeyValue<>(record.key(), record.value())); - } - } - while (count < expectedClicksPerRegion.size() - && (System.currentTimeMillis() - start) < 30000); - - // TODO: Matched count is 3 and not 4 (expectedClicksPerRegion.size()) when running with full suite. Investigate why. - // TODO: This behavior is only observed after the Spring Kafka upgrade to 2.5.0 and kafka client to 2.5. - // TODO: Note that the test passes fine as a single test. - assertThat(count).matches( - matchedCount -> matchedCount == expectedClicksPerRegion.size() - 1 || matchedCount == expectedClicksPerRegion.size()); - assertThat(actualClicksPerRegion).containsAnyElementsOf(expectedClicksPerRegion); - } - finally { - consumer.close(); - } - } - - @Test - public void testTrivialSingleKTableInputAsNonDeclarative() { - SpringApplication app = new SpringApplication( - TrivialKTableApp.class); - app.setWebApplicationType(WebApplicationType.NONE); - app.run("--server.port=0", - "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString(), - "--spring.cloud.stream.kafka.streams.bindings.input-y.consumer.application-id=" + - "testTrivialSingleKTableInputAsNonDeclarative"); - //All we are verifying is that this application didn't throw any errors. - //See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/536 - } - - @Test - public void testTwoKStreamsCanBeJoined() { - SpringApplication app = new SpringApplication( - JoinProcessor.class); - app.setWebApplicationType(WebApplicationType.NONE); - app.run("--server.port=0", - "--spring.cloud.stream.kafka.streams.binder.brokers=" - + embeddedKafka.getBrokersAsString(), - "--spring.application.name=" + - "two-kstream-input-join-integ-test"); - //All we are verifying is that this application didn't throw any errors. - //See this issue: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/701 - } - - @EnableBinding(KafkaStreamsProcessorX.class) - @EnableAutoConfiguration - public static class CountClicksPerRegionApplication { - - @StreamListener - @SendTo("output") - public KStream process( - @Input("input") KStream userClicksStream, - @Input("input-x") KTable userRegionsTable) { - - return userClicksStream - .leftJoin(userRegionsTable, - (clicks, region) -> new RegionWithClicks( - region == null ? "UNKNOWN" : region, clicks), - Joined.with(Serdes.String(), Serdes.Long(), null)) - .map((user, regionWithClicks) -> new KeyValue<>( - regionWithClicks.getRegion(), regionWithClicks.getClicks())) - .groupByKey(Serialized.with(Serdes.String(), Serdes.Long())) - .reduce(Long::sum) - .toStream(); - } - - //This forces the state stores to be cleaned up before running the test. - @Bean - public CleanupConfig cleanupConfig() { - return new CleanupConfig(true, false); - } - - } - - @EnableBinding(KafkaStreamsProcessorY.class) - @EnableAutoConfiguration - public static class TrivialKTableApp { - - @StreamListener("input-y") - public void process(KTable inputTable) { - inputTable.toStream().foreach((key, value) -> System.out.println("key : value " + key + " : " + value)); - } - } - - interface KafkaStreamsProcessorX extends KafkaStreamsProcessor { - - @Input("input-x") - KTable inputX(); - - } - - interface KafkaStreamsProcessorY { - - @Input("input-y") - KTable inputY(); - - } - - /** - * Tuple for a region and its associated number of clicks. - */ - private static final class RegionWithClicks { - - private final String region; - - private final long clicks; - - RegionWithClicks(String region, long clicks) { - if (region == null || region.isEmpty()) { - throw new IllegalArgumentException("region must be set"); - } - if (clicks < 0) { - throw new IllegalArgumentException("clicks must not be negative"); - } - this.region = region; - this.clicks = clicks; - } - - public String getRegion() { - return region; - } - - public long getClicks() { - return clicks; - } - - } - - interface BindingsForTwoKStreamJoinTest { - - String INPUT_1 = "input_1"; - String INPUT_2 = "input_2"; - - @Input(INPUT_1) - KStream input_1(); - - @Input(INPUT_2) - KStream input_2(); - } - - @EnableBinding(BindingsForTwoKStreamJoinTest.class) - @EnableAutoConfiguration - public static class JoinProcessor { - - @StreamListener - public void testProcessor( - @Input(BindingsForTwoKStreamJoinTest.INPUT_1) KStream input1Stream, - @Input(BindingsForTwoKStreamJoinTest.INPUT_2) KStream input2Stream) { - input1Stream - .join(input2Stream, - (event1, event2) -> null, - JoinWindows.of(TimeUnit.MINUTES.toMillis(5)), - Joined.with( - Serdes.String(), - Serdes.String(), - Serdes.String() - ) - ); - } - } - -}