From 3668a70b197b8c9f62ff735e3df031e6146bc370 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sun, 9 Jun 2024 11:06:20 +0200 Subject: [PATCH] query: 1.0.6 - Refined `QueryStreamProcessor` -- moved the building of the toplogy into a separate (static) method --- .../wordcount/query/QueryStreamProcessor.java | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java index 15b28db..efa28ca 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java @@ -6,10 +6,7 @@ import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.common.serialization.Serdes; -import org.apache.kafka.streams.KafkaStreams; -import org.apache.kafka.streams.KeyQueryMetadata; -import org.apache.kafka.streams.StoreQueryParameters; -import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.*; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; @@ -39,6 +36,18 @@ public class QueryStreamProcessor String usersInputTopic, String rankingInputTopic, ObjectMapper mapper) + { + Topology topology = buildTopology(usersInputTopic, rankingInputTopic, mapper); + streams = new KafkaStreams(topology, props); + hostInfo = applicationServer; + storeParameters = StoreQueryParameters.fromNameAndType(STORE_NAME, QueryableStoreTypes.keyValueStore());; + this.mapper = mapper; + } + + static Topology buildTopology( + String usersInputTopic, + String rankingInputTopic, + ObjectMapper mapper) { StreamsBuilder builder = new StreamsBuilder(); @@ -66,10 +75,10 @@ public class QueryStreamProcessor }) .toTable(Materialized.as(STORE_NAME)); - streams = new KafkaStreams(builder.build(), props); - hostInfo = applicationServer; - storeParameters = StoreQueryParameters.fromNameAndType(STORE_NAME, QueryableStoreTypes.keyValueStore());; - this.mapper = mapper; + Topology topology = builder.build(); + log.info("\n\n{}", topology.describe()); + + return topology; } public Optional getRedirect(String username) -- 2.20.1