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

Add KafkaStream interactive query sample app (#2461)

Browse Source 
* Add KafkaStream interactive query sample app

* Remove KafkaStreams interactive query "basic" sample app
This commit is contained in:
Chris Bono
2022-07-25 17:07:59 -05:00
committed by GitHub
parent f1a0e76e80
commit 482b945bfd
12 changed files with 461 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/GenerateProducts.java Normal file
View File

@@ -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<Product> productSerde = new JsonSerde<>(Product.class);
Map<String, Object> 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<Integer, Product> producerFactory = new DefaultKafkaProducerFactory<>(props);
KafkaTemplate<Integer, Product> template = new KafkaTemplate<>(producerFactory, true);
template.setDefaultTopic(InteractiveProductCountApplication.PRODUCT_TOPIC);
final List<Product> 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);
}
}
}

183
samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/KafkaStreamsInteractiveQueryApplication.java Normal file
View File

@@ -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<KStream<Object, Product>> process() {
return (input) -> {
// Accept product events that are configured to be tracked
final KStream<Integer, Product> productsByProductId = input
.filter((key, product) -> productTrackerProperties.getProductIds().contains(product.id()))
.map((key, value) -> new KeyValue<>(value.id, value));
Serde<Product> productSerde = new JsonSerde<>(Product.class);
Serde<ProductCount> productCountSerde = new JsonSerde<>(ProductCount.class);
Serde<TopTwoProducts> topTwoSerde = new JsonSerde<>(TopTwoProducts.class).noTypeInfo();
// Create a state store to track product counts
final KTable<Product, Long> productCounts = productsByProductId
.groupBy((productId, product) -> product, Grouped.with(productSerde, productSerde))
.count(Materialized.<Product, Long, KeyValueStore<Bytes, byte[]>>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.<String, TopTwoProducts, KeyValueStore<Bytes, byte[]>>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<Integer> productIds = new HashSet<>();
public Set<Integer> getProductIds() {
return productIds;
}
public void setProductIds(Set<Integer> productIds) {
this.productIds = productIds;
}
}
/**
* Used in aggregations to keep track of the Top two products
*/
static class TopTwoProducts {
@JsonIgnore
private final Map<Integer, ProductCount> currentProducts = new HashMap<>();
@JsonIgnore
private final TreeSet<ProductCount> 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<ProductCount> products() {
return topTwo.stream().toList();
}
@JsonSetter
@SuppressWarnings("unused")
public void setProducts(List<ProductCount> products) {
products.forEach(this::add);
}
}
}

83
samples/kafka-streams-interactive-query/src/main/java/com/example/kafka/streams/music/ProductQueryController.java Normal file
View File

@@ -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> productCount(@PathVariable("id") Integer productId) {
JsonSerializer<Product> productSerializer = new JsonSerde<Product>().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<Product, Long> 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<List<ProductCount>> 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<String, TopTwoProducts> 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<>() {});
}
}

15
samples/kafka-streams-interactive-query/src/main/resources/application.yml Normal file
View File

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