X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Ftop10%2FTop10StreamProcessor.java;h=2ff078c181225a87251a50bdfcd8f5dceb8a355e;hb=b69e1270a308f200b2640a01d37d4636a0a549e1;hp=d3846d89beab0cb573ddca8dcc4ffc477afee2fd;hpb=df0c22234e9ace115b4e4abc4e3d881d5595668e;p=demos%2Fkafka%2Fwordcount diff --git a/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java index d3846d8..2ff078c 100644 --- a/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/top10/Top10StreamProcessor.java @@ -5,6 +5,8 @@ import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.KeyValue; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.Topology; +import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; +import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; import java.util.Properties; @@ -22,14 +24,16 @@ public class Top10StreamProcessor { Topology topology = Top10StreamProcessor.buildTopology( inputTopic, - outputTopic); + outputTopic, + null); streams = new KafkaStreams(topology, props); } static Topology buildTopology( String inputTopic, - String outputTopic) + String outputTopic, + KeyValueBytesStoreSupplier storeSupplier) { StreamsBuilder builder = new StreamsBuilder(); @@ -49,6 +53,11 @@ public class Top10StreamProcessor return topology; } + ReadOnlyKeyValueStore getStore(String name) + { + return null; + } + public void start() { log.info("Starting Stream-Processor");