X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2Fwordcount%2Fcounter%2FCounterStreamProcessor.java;h=b1343a775e183e91f543ed2ec8c9986d5b8def91;hb=35943a021ac4b25fad29861ecc083b6974ea2732;hp=4cc0c6866bb7d31613f179debd285f3f3d9c9566;hpb=f58f069e556e9782656cf4e1d09eabbfcd6fa1c4;p=demos%2Fkafka%2Fwordcount diff --git a/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java index 4cc0c68..b1343a7 100644 --- a/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/counter/CounterStreamProcessor.java @@ -1,17 +1,15 @@ package de.juplo.kafka.wordcount.counter; -import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.common.serialization.Serdes; 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.kstream.Consumed; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; -import org.apache.kafka.streams.kstream.*; -import org.springframework.kafka.support.serializer.JsonSerde; import java.util.Properties; @@ -26,14 +24,12 @@ public class CounterStreamProcessor String inputTopic, String outputTopic, Properties properties, - KeyValueBytesStoreSupplier storeSupplier, - ObjectMapper mapper) + KeyValueBytesStoreSupplier storeSupplier) { Topology topology = CounterStreamProcessor.buildTopology( inputTopic, outputTopic, - storeSupplier, - mapper); + storeSupplier); streams = new KafkaStreams(topology, properties); } @@ -41,37 +37,21 @@ public class CounterStreamProcessor static Topology buildTopology( String inputTopic, String outputTopic, - KeyValueBytesStoreSupplier storeSupplier, - ObjectMapper mapper) + KeyValueBytesStoreSupplier storeSupplier) { StreamsBuilder builder = new StreamsBuilder(); KStream source = builder.stream( inputTopic, - Consumed.with( - Serdes.String(), - new JsonSerde<>(Word.class) - .ignoreTypeHeaders())); + Consumed.with(Serdes.String(), null)); source .map((key, word) -> new KeyValue<>(word, word)) - .groupByKey(Grouped.with( - new JsonSerde<>(Word.class) - .forKeys() - .noTypeInfo(), - new JsonSerde<>(Word.class) - .noTypeInfo())) + .groupByKey() .count(Materialized.as(storeSupplier)) .toStream() - .map((word, count) -> new KeyValue<>(word, WordCount.of(word, count))) - .to( - outputTopic, - Produced.with( - new JsonSerde<>(Word.class) - .forKeys() - .noTypeInfo(), - new JsonSerde<>(WordCount.class) - .noTypeInfo())); + .map((word, counter) -> new KeyValue<>(word, WordCounter.of(word, counter))) + .to(outputTopic); Topology topology = builder.build(); log.info("\n\n{}", topology.describe());