From 482b945bfd53b9bb2ba6f2416e6429b11481e4fc Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 25 Jul 2022 17:07:59 -0500 Subject: [PATCH] Add KafkaStream interactive query sample app (#2461) * Add KafkaStream interactive query sample app * Remove KafkaStreams interactive query "basic" sample app --- .../README.adoc | 63 ------ ...fkaStreamsInteractiveQueryApplication.java | 113 ----------- .../src/main/resources/application.yml | 12 -- .../README.adoc | 101 ++++++++++ .../mvnw | 0 .../mvnw.cmd | 0 .../pom.xml | 4 +- .../kafka/streams/music/GenerateProducts.java | 76 ++++++++ ...fkaStreamsInteractiveQueryApplication.java | 183 ++++++++++++++++++ .../streams/music/ProductQueryController.java | 83 ++++++++ .../src/main/resources/application.yml | 15 ++ samples/pom.xml | 2 +- 12 files changed, 461 insertions(+), 191 deletions(-) delete mode 100644 samples/kafka-streams-interactive-query-basic/README.adoc delete mode 100644 samples/kafka-streams-interactive-query-basic/src/main/java/kafka/streams/product/tracker/KafkaStreamsInteractiveQueryApplication.java delete mode 100644 samples/kafka-streams-interactive-query-basic/src/main/resources/application.yml create mode 100644 samples/kafka-streams-interactive-query/README.adoc rename samples/{kafka-streams-interactive-query-basic => kafka-streams-interactive-query}/mvnw (100%) rename samples/{kafka-streams-interactive-query-basic => kafka-streams-interactive-query}/mvnw.cmd (100%) rename samples/{kafka-streams-interactive-query-basic => kafka-streams-interactive-query}/pom.xml (90%) create mode 100644 samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/GenerateProducts.java create mode 100644 samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/KafkaStreamsInteractiveQueryApplication.java create mode 100644 samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/ProductQueryController.java create mode 100644 samples/kafka-streams-interactive-query/src/main/resources/application.yml diff --git a/samples/kafka-streams-interactive-query-basic/README.adoc b/samples/kafka-streams-interactive-query-basic/README.adoc deleted file mode 100644 index bdadc2c41..000000000 --- a/samples/kafka-streams-interactive-query-basic/README.adoc +++ /dev/null @@ -1,63 +0,0 @@ -== Spring Cloud Stream Kafka Streams Interactive Query (basic) - -This sample demonstrates a Spring Cloud Stream processor using Kafka Streams basic state store query support. - -=== Application -The app is based on a contrived use case of tracking products by interactively querying their status. The program accepts product ID's and tracks their counts hitherto by interactively querying the underlying store. - -[[build-app]] -=== Building -To build the app simply execute the following command: -[source,bash] ----- -./mvnw clean install ----- - -=== Running - -==== Ensure these pre-requisites -**** -* The app has been built by following the <> steps -* Apache Kafka broker available at `localhost:9092` - -[#kafka_tools] -TIP: The included xref:../../../tools/kafka/docker-compose/README.adoc#run_kafka_cluster[Kafka tools] can be used to easily start a broker at the required coordinates -**** - -==== Start the streams app -[source,bash] ----- -java -jar target/kafka-streams-interactive-query-basic-4.0.0-SNAPSHOT.jar --app.product.tracker.product-ids=123,124,125 ----- -The above command will track products with ID's `123`,`124` and `125` and print their counts seen so far every 30 seconds. - -===== Send input messages -Leverage the Kafka command line tool `kafka-console-producer` to send messges (products) to the input topic. - -Issue the following command on a separate terminal: - -[source,bash] ----- -docker exec -it broker1 /bin/kafka-console-producer --broker-list broker1:29091 --topic products ----- -NOTE: The command reference `broker1` and `broker1:29091` as they assume the aforementioned <> are used to create the cluster. If you start up your own cluster you will need to adjust the coordinates accordingly. Also note that the port is `29091` as that is the internal port configured on the cluster (rather than the expected `9091`) - -Enter the following in the console producer (one line at a time) and watch the output on the console (or IDE) where the application is running. - -[source,bash] ----- -{"id":"123"} -{"id":"124"} -{"id":"125"} -{"id":"123"} -{"id":"123"} -{"id":"123"} ----- - -The output should look something like the following: -[source,bash] ----- -Product ID: 123 Count: 4 -Product ID: 124 Count: 1 -Product ID: 125 Count: 1 ----- diff --git a/samples/kafka-streams-interactive-query-basic/src/main/java/kafka/streams/product/tracker/KafkaStreamsInteractiveQueryApplication.java b/samples/kafka-streams-interactive-query-basic/src/main/java/kafka/streams/product/tracker/KafkaStreamsInteractiveQueryApplication.java deleted file mode 100644 index f85ac75ec..000000000 --- a/samples/kafka-streams-interactive-query-basic/src/main/java/kafka/streams/product/tracker/KafkaStreamsInteractiveQueryApplication.java +++ /dev/null @@ -1,113 +0,0 @@ -/* - * Copyright 2017-2022 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 kafka.streams.product.tracker; - -import java.util.HashSet; -import java.util.Set; -import java.util.function.Function; - -import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.common.utils.Bytes; -import org.apache.kafka.streams.KeyValue; -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.state.KeyValueStore; -import org.apache.kafka.streams.state.QueryableStoreTypes; -import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import org.springframework.boot.SpringApplication; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService; -import org.springframework.context.annotation.Bean; -import org.springframework.kafka.support.serializer.JsonSerde; -import org.springframework.scheduling.annotation.EnableScheduling; -import org.springframework.scheduling.annotation.Scheduled; - -@SpringBootApplication(proxyBeanMethods = false) -class KafkaStreamsInteractiveQueryApplication { - - public static void main(String[] args) { - SpringApplication.run(KafkaStreamsInteractiveQueryApplication.class, args); - } - - @EnableConfigurationProperties(ProductTrackerProperties.class) - @EnableScheduling - static class InteractiveProductCountApplication { - private static final Logger LOG = LoggerFactory.getLogger(InteractiveProductCountApplication.class); - private static final String STORE_NAME = "prod-id-count-store"; - private final InteractiveQueryService queryService; - private final ProductTrackerProperties productTrackerProperties; - private ReadOnlyKeyValueStore keyValueStore; - - InteractiveProductCountApplication(InteractiveQueryService queryService, ProductTrackerProperties productTrackerProperties) { - this.queryService = queryService; - this.productTrackerProperties = productTrackerProperties; - } - - @Bean - public Function, KStream> process() { - return input -> input - .filter((key, product) -> productTrackerProperties.getProductIds().contains(product.getId())) - .map((key, value) -> new KeyValue<>(value.id, value)) - .groupByKey(Grouped.with(Serdes.Integer(), new JsonSerde<>(Product.class))) - .count(Materialized.>as(STORE_NAME) - .withKeySerde(Serdes.Integer()) - .withValueSerde(Serdes.Long())) - .toStream(); - } - - @Scheduled(fixedRate = 30000, initialDelay = 5000) - void logProductCounts() { - if (keyValueStore == null) { - keyValueStore = queryService.getQueryableStore(STORE_NAME, QueryableStoreTypes.keyValueStore()); - } - productTrackerProperties.getProductIds().forEach((id) -> LOG.info("Product ID: " + id + " Count: " + keyValueStore.get(id))); - } - } - - @ConfigurationProperties(prefix = "app.product.tracker") - static class ProductTrackerProperties { - - private Set productIds = new HashSet<>(); - - public Set getProductIds() { - return productIds; - } - - public void setProductIds(Set productIds) { - this.productIds = productIds; - } - } - - static class Product { - - private Integer id; - - public Integer getId() { - return id; - } - - public void setId(Integer id) { - this.id = id; - } - } -} diff --git a/samples/kafka-streams-interactive-query-basic/src/main/resources/application.yml b/samples/kafka-streams-interactive-query-basic/src/main/resources/application.yml deleted file mode 100644 index 6dab51339..000000000 --- a/samples/kafka-streams-interactive-query-basic/src/main/resources/application.yml +++ /dev/null @@ -1,12 +0,0 @@ -spring: - application.name: kafka-streams-iq-basic-sample - cloud: - stream: - kafka.streams.binder: - configuration: - commit.interval.ms: 1000 - bindings: - process-in-0: - destination: products - process-out-0: - destination: product-counts diff --git a/samples/kafka-streams-interactive-query/README.adoc b/samples/kafka-streams-interactive-query/README.adoc new file mode 100644 index 000000000..86367d049 --- /dev/null +++ b/samples/kafka-streams-interactive-query/README.adoc @@ -0,0 +1,101 @@ +== Spring Cloud Stream Kafka Streams Interactive Query + +This sample demonstrates a Spring Cloud Stream processor using Kafka Streams state store interactive query support. There is a REST service provided as part of the application that can be used to query the store interactively. + +=== Application + +The app is based on a contrived use case of tracking products by interactively querying their status. The program accepts product ID's and tracks their counts (including the top two) and makes the results available by interactively querying the underlying state stores. + +[[build-app]] +=== Building +To build the app simply execute the following command in the base directory: +[source,bash] +---- +./mvnw clean install +---- + +=== Running + +==== Ensure these pre-requisites +**** +* The app has been built by following the <> steps +* Apache Kafka broker available at `localhost:9092` + +[#kafka_tools] +TIP: The included xref:../../../tools/kafka/docker-compose/README.adoc#run_kafka_cluster[Kafka tools] can be used to easily start a broker at the required coordinates +**** + +We will run 2 instances of the app to demonstrate that regardless of which instance hosts the keys, the REST endpoint will serve the requests. + +NOTE: For more information on how this is done, please take a look at the link:./src/main/java/com/example/kafka/streams/music/ProductQueryController.java[ProductQueryController] + +==== Start the streams app (instance 1) +[source,bash] +---- +java -jar target/kafka-streams-interactive-query-advanced-4.0.0-SNAPSHOT.jar --app.product.tracker.product-ids=123,124,125 +---- + +==== Start the streams app (instance 2) +[source,bash] +---- +java -jar target/kafka-streams-interactive-query-advanced-4.0.0-SNAPSHOT.jar --server.port=8082 --spring.cloud.stream.kafka.streams.binder.configuration.application.server=localhost:8082 --app.product.tracker.product-ids=123,124,125 +---- + +==== Start the data generator +Run the stand-alone link:./src/main/java/com/example/kafka/streams/music/GenerateProducts.java[GenerateProducts] app to create random input data. + +NOTE: It can easily be started from within your IDE or on the command line with the following command: + +[source,bash] +---- +./mvnw exec:java -Dexec.mainClass="com.example.kafka.streams.music.GenerateProducts" +---- + +The output should look something like the following: +[source,bash] +---- +GenerateProducts - Writing product event for productId = 124 +GenerateProducts - Writing product event for productId = 127 +GenerateProducts - Writing product event for productId = 127 +GenerateProducts - Writing product event for productId = 124 +---- + +==== Query the state stores + +===== Product count +Navigate to http://localhost:8080/product/123 to view the count for a particular product id. Keep refreshing the URL and you will see the product count increase as the generator runs. + +Now navigate to http://localhost:8082/product/123 to view the count for the same product but served from the second app instance. Keep refreshing the URL and you will see the product count increase as the generator runs. + +Take a look at the console sessions for the applications and you will notice that one of the instances actually serves the count from its state store and the other instance simply proxies the request to that instance to retrieve the count info. + +For example, the first app instance logs may look like: +[source,bash,options=nowrap,subs=attributes] +---- +Product count for productId: 123 served from different host: HostInfo{host='localhost', port=8082} +20 +---- +and the second app instance logs may look like: +[source,bash,options=nowrap,subs=attributes] +---- +Product count for productId: 123 served from same host: HostInfo{host='localhost', port=8082} +---- + +===== Top-two products + +Navigate to http://localhost:8080/product/top-two to view the two products with the highest count at the time of the query. Keep refreshing the URL and you may see the top-two products change as the generator runs. + +Now navigate to http://localhost:8082/product/top-two to view the top-two products but served from the second app instance. Keep refreshing the URL and you may see the top-two products change as the generator runs. + +Take a look at the console sessions for the applications and you will notice that one of the instances actually serves the results from its state store and the other instance simply proxies the request to that instance to retrieve the top-two info. + +For example, the first app instance logs may look like: +[source,bash,options=nowrap,subs=attributes] +---- +Top two products served from different host: HostInfo{host='localhost', port=8082} +---- +and the second app instance logs may look like: +[source,bash,options=nowrap,subs=attributes] +---- +Top two products served from same host: HostInfo{host='localhost', port=8082} +---- diff --git a/samples/kafka-streams-interactive-query-basic/mvnw b/samples/kafka-streams-interactive-query/mvnw similarity index 100% rename from samples/kafka-streams-interactive-query-basic/mvnw rename to samples/kafka-streams-interactive-query/mvnw diff --git a/samples/kafka-streams-interactive-query-basic/mvnw.cmd b/samples/kafka-streams-interactive-query/mvnw.cmd similarity index 100% rename from samples/kafka-streams-interactive-query-basic/mvnw.cmd rename to samples/kafka-streams-interactive-query/mvnw.cmd diff --git a/samples/kafka-streams-interactive-query-basic/pom.xml b/samples/kafka-streams-interactive-query/pom.xml similarity index 90% rename from samples/kafka-streams-interactive-query-basic/pom.xml rename to samples/kafka-streams-interactive-query/pom.xml index 9b447a1ef..1c163f23c 100644 --- a/samples/kafka-streams-interactive-query-basic/pom.xml +++ b/samples/kafka-streams-interactive-query/pom.xml @@ -3,8 +3,8 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 com.example - kafka-streams-interactive-query-basic - kafka-streams-interactive-query-basic + kafka-streams-interactive-query-advanced + kafka-streams-interactive-query-advanced org.springframework.cloud diff --git a/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/GenerateProducts.java b/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/GenerateProducts.java new file mode 100644 index 000000000..fdc73afb2 --- /dev/null +++ b/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/GenerateProducts.java @@ -0,0 +1,76 @@ +/* + * Copyright 2018-2022 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 com.example.kafka.streams.music; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; + +import com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.InteractiveProductCountApplication; +import com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.Product; +import org.apache.kafka.clients.producer.ProducerConfig; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.Serde; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.support.serializer.JsonSerde; + +public class GenerateProducts { + + private static final Logger LOG = LoggerFactory.getLogger(GenerateProducts.class); + + public static void main(String... args) throws Exception { + + Serde productSerde = new JsonSerde<>(Product.class); + + Map props = new HashMap<>(); + props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); + props.put(ProducerConfig.RETRIES_CONFIG, 0); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); + props.put(ProducerConfig.LINGER_MS_CONFIG, 1); + props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); + props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class); + props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, productSerde.serializer().getClass()); + + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(props); + KafkaTemplate template = new KafkaTemplate<>(producerFactory, true); + template.setDefaultTopic(InteractiveProductCountApplication.PRODUCT_TOPIC); + + final List products = Arrays.asList( + new Product(123), + new Product(124), + new Product(125), + new Product(126), + new Product(127)); + + final Random random = new Random(); + + // send a product event every 100 milliseconds + while (true) { + final Product product = products.get(random.nextInt(products.size())); + LOG.debug("Writing product event for productId = {}", product.id()); + template.sendDefault(product.id(), product); + Thread.sleep(1000L); + } + + } +} diff --git a/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/KafkaStreamsInteractiveQueryApplication.java b/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/KafkaStreamsInteractiveQueryApplication.java new file mode 100644 index 000000000..74cb6ec37 --- /dev/null +++ b/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/KafkaStreamsInteractiveQueryApplication.java @@ -0,0 +1,183 @@ +/* + * Copyright 2017-2022 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 com.example.kafka.streams.music; + +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.function.Consumer; + +import javax.annotation.PostConstruct; + +import com.fasterxml.jackson.annotation.JsonGetter; +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonSetter; +import org.apache.kafka.common.serialization.Serde; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.Bytes; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Grouped; +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.state.KeyValueStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.support.serializer.JsonSerde; + +@SpringBootApplication(proxyBeanMethods = false) +class KafkaStreamsInteractiveQueryApplication { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsInteractiveQueryApplication.class); + + public static void main(String[] args) { + SpringApplication.run(KafkaStreamsInteractiveQueryApplication.class, args); + } + + @EnableConfigurationProperties(ProductTrackerProperties.class) + static class InteractiveProductCountApplication { + + static final String PRODUCT_TOPIC = "product-events"; + static final String PRODUCT_COUNT_STORE_NAME = "product-count-store"; + static final String TOP_TWO_PRODUCTS_STORE_NAME = "top-two-products-store"; + static final String TOP_TWO_PRODUCTS_KEY = "top-two"; + + @Value("${spring.cloud.stream.kafka.streams.binder.configuration.state.dir:UNKNOWN}") + private String stateStoreDir; + + private final ProductTrackerProperties productTrackerProperties; + + InteractiveProductCountApplication(ProductTrackerProperties productTrackerProperties) { + this.productTrackerProperties = productTrackerProperties; + } + + @PostConstruct + void showStateStoreInfo() { + LOG.info("Using KafkaStreams state.dir = {}", this.stateStoreDir); + } + + @Bean + public Consumer> process() { + return (input) -> { + // Accept product events that are configured to be tracked + final KStream productsByProductId = input + .filter((key, product) -> productTrackerProperties.getProductIds().contains(product.id())) + .map((key, value) -> new KeyValue<>(value.id, value)); + + Serde productSerde = new JsonSerde<>(Product.class); + Serde productCountSerde = new JsonSerde<>(ProductCount.class); + Serde topTwoSerde = new JsonSerde<>(TopTwoProducts.class).noTypeInfo(); + + // Create a state store to track product counts + final KTable productCounts = productsByProductId + .groupBy((productId, product) -> product, Grouped.with(productSerde, productSerde)) + .count(Materialized.>as(PRODUCT_COUNT_STORE_NAME) + .withKeySerde(productSerde) + .withValueSerde(Serdes.Long())); + + // Compute the top two products and store updated result in 'top-two-products-store' + productCounts.groupBy((product, count) -> + KeyValue.pair(TOP_TWO_PRODUCTS_KEY, new ProductCount(product.id(), count)), + Grouped.with(Serdes.String(), productCountSerde)) + .aggregate( + TopTwoProducts::new, + (aggKey, value, aggregate) -> aggregate.add(value), + (aggKey, value, aggregate) -> aggregate.remove(value), + Materialized.>as(TOP_TWO_PRODUCTS_STORE_NAME) + .withKeySerde(Serdes.String()) + .withValueSerde(topTwoSerde)); //new TopTwoProductsSerde())); + }; + } + } + + record Product(Integer id) { } + + record ProductCount(Integer productId, Long count) { } + + @ConfigurationProperties(prefix = "app.product.tracker") + static class ProductTrackerProperties { + + private Set productIds = new HashSet<>(); + + public Set getProductIds() { + return productIds; + } + + public void setProductIds(Set productIds) { + this.productIds = productIds; + } + } + + /** + * Used in aggregations to keep track of the Top two products + */ + static class TopTwoProducts { + + @JsonIgnore + private final Map currentProducts = new HashMap<>(); + + @JsonIgnore + private final TreeSet topTwo = new TreeSet<>((o1, o2) -> { + final int result = Long.compare(o2.count(), o1.count()); + if (result != 0) { + return result; + } + return Integer.compare(o1.productId(), o2.productId()); + }); + + public TopTwoProducts add(final ProductCount productCount) { + if (currentProducts.containsKey(productCount.productId())) { + topTwo.remove(currentProducts.remove(productCount.productId())); + } + topTwo.add(productCount); + currentProducts.put(productCount.productId(), productCount); + if (topTwo.size() > 2) { + final ProductCount last = topTwo.last(); + currentProducts.remove(last.productId()); + topTwo.remove(last); + } + return this; + } + + public TopTwoProducts remove(final ProductCount value) { + topTwo.remove(value); + currentProducts.remove(value.productId()); + return this; + } + + @JsonGetter + public List products() { + return topTwo.stream().toList(); + } + + @JsonSetter + @SuppressWarnings("unused") + public void setProducts(List products) { + products.forEach(this::add); + } + } +} diff --git a/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/ProductQueryController.java b/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/ProductQueryController.java new file mode 100644 index 000000000..0ace02a20 --- /dev/null +++ b/samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/ProductQueryController.java @@ -0,0 +1,83 @@ +package com.example.kafka.streams.music; + +import java.util.List; +import java.util.Optional; + +import com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.Product; +import com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.ProductCount; +import com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.TopTwoProducts; +import org.apache.kafka.common.serialization.StringSerializer; +import org.apache.kafka.streams.state.HostInfo; +import org.apache.kafka.streams.state.QueryableStoreTypes; +import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService; +import org.springframework.core.ParameterizedTypeReference; +import org.springframework.http.HttpMethod; +import org.springframework.http.ResponseEntity; +import org.springframework.kafka.support.serializer.JsonSerde; +import org.springframework.kafka.support.serializer.JsonSerializer; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.client.RestTemplate; + +import static com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.InteractiveProductCountApplication.PRODUCT_COUNT_STORE_NAME; +import static com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.InteractiveProductCountApplication.TOP_TWO_PRODUCTS_KEY; +import static com.example.kafka.streams.music.KafkaStreamsInteractiveQueryApplication.InteractiveProductCountApplication.TOP_TWO_PRODUCTS_STORE_NAME; + +@RestController +class ProductQueryController { + + private static final Logger LOG = LoggerFactory.getLogger(ProductQueryController.class); + + private final InteractiveQueryService queryService; + + ProductQueryController(InteractiveQueryService queryService) { + this.queryService = queryService; + } + + @GetMapping("/product/{id}") + public ResponseEntity productCount(@PathVariable("id") Integer productId) { + JsonSerializer productSerializer = new JsonSerde().serializer(); + HostInfo hostInfo = queryService.getHostInfo(PRODUCT_COUNT_STORE_NAME, new Product(productId), productSerializer); + if (queryService.getCurrentHostInfo().equals(hostInfo)) { + LOG.info("Product count for productId: {} served from same host: {}", productId, hostInfo); + ReadOnlyKeyValueStore productCountStore = + queryService.getQueryableStore(PRODUCT_COUNT_STORE_NAME, QueryableStoreTypes.keyValueStore()); + Long count = productCountStore.get(new Product(productId)); + if (count == null) { + return ResponseEntity.notFound().build(); + } + return ResponseEntity.of(Optional.of(new ProductCount(productId, count))); + } + LOG.info("Product count for productId: {} served from different host: {}", productId, hostInfo); + RestTemplate restTemplate = new RestTemplate(); + return restTemplate.getForEntity( + String.format("http://%s:%d/product/%d", hostInfo.host(), hostInfo.port(), productId), ProductCount.class); + } + + @GetMapping("/product/top-two") + public ResponseEntity> topTwo() { + HostInfo hostInfo = queryService.getHostInfo(TOP_TWO_PRODUCTS_STORE_NAME, TOP_TWO_PRODUCTS_KEY, new StringSerializer()); + if (queryService.getCurrentHostInfo().equals(hostInfo)) { + LOG.info("Top two products served from same host: {}", hostInfo); + ReadOnlyKeyValueStore topTwoProductsStore = + queryService.getQueryableStore(TOP_TWO_PRODUCTS_STORE_NAME, QueryableStoreTypes.keyValueStore()); + TopTwoProducts topTwo = topTwoProductsStore.get(TOP_TWO_PRODUCTS_KEY); + if (topTwo == null) { + return ResponseEntity.notFound().build(); + } + return ResponseEntity.of(Optional.of(topTwo.products())); + } + LOG.info("Top two products served from different host: {}", hostInfo); + RestTemplate restTemplate = new RestTemplate(); + return restTemplate.exchange(String.format("http://%s:%d/product/top-two", hostInfo.host(), hostInfo.port()), + HttpMethod.GET, + null, + new ParameterizedTypeReference<>() {}); + } + +} diff --git a/samples/kafka-streams-interactive-query/src/main/resources/application.yml b/samples/kafka-streams-interactive-query/src/main/resources/application.yml new file mode 100644 index 000000000..138420476 --- /dev/null +++ b/samples/kafka-streams-interactive-query/src/main/resources/application.yml @@ -0,0 +1,15 @@ +spring: + application.name: kafka-streams-iq-adv-sample + cloud: + stream: + bindings: + process-in-0: + destination: product-events + kafka.streams: + binder: + auto-add-partitions: true + min-partition-count: 4 + configuration: + application.server: localhost:8080 + commit.interval.ms: 1000 + state.dir: "${java.io.tmpdir}/kafka-streams_${server.port:0}" diff --git a/samples/pom.xml b/samples/pom.xml index 1926b28d2..60319bca0 100644 --- a/samples/pom.xml +++ b/samples/pom.xml @@ -25,7 +25,7 @@ confluent-schema-registry-integration kafka-native-serialization kafka-streams-branching - kafka-streams-interactive-query-basic + kafka-streams-interactive-query stream-bridge-avro