counter: 1.3.1 - Splitted up test in `CounterStreamProcessorTopologyTest`
[demos/kafka/wordcount] / src / main / java / de / juplo / kafka / wordcount / counter / CounterStreamProcessor.java
1 package de.juplo.kafka.wordcount.counter;
2
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.common.serialization.Serdes;
5 import org.apache.kafka.streams.KafkaStreams;
6 import org.apache.kafka.streams.KeyValue;
7 import org.apache.kafka.streams.StreamsBuilder;
8 import org.apache.kafka.streams.Topology;
9 import org.apache.kafka.streams.kstream.Consumed;
10 import org.apache.kafka.streams.kstream.KStream;
11 import org.apache.kafka.streams.kstream.Materialized;
12 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
13
14 import java.util.Properties;
15
16
17 @Slf4j
18 public class CounterStreamProcessor
19 {
20         public final KafkaStreams streams;
21
22
23         public CounterStreamProcessor(
24                         String inputTopic,
25                         String outputTopic,
26                         Properties properties,
27                         KeyValueBytesStoreSupplier storeSupplier)
28         {
29                 Topology topology = CounterStreamProcessor.buildTopology(
30                                 inputTopic,
31                                 outputTopic,
32                                 storeSupplier);
33
34                 streams = new KafkaStreams(topology, properties);
35         }
36
37         static Topology buildTopology(
38                         String inputTopic,
39                         String outputTopic,
40                         KeyValueBytesStoreSupplier storeSupplier)
41         {
42                 StreamsBuilder builder = new StreamsBuilder();
43
44                 KStream<String, Word> source = builder.stream(
45                                 inputTopic,
46                                 Consumed.with(Serdes.String(), null));
47
48                 source
49                                 .map((key, word) -> new KeyValue<>(word, word))
50                                 .groupByKey()
51                                 .count(Materialized.as(storeSupplier))
52                                 .toStream()
53                                 .map((word, counter) -> new KeyValue<>(word, WordCounter.of(word, counter)))
54                                 .to(outputTopic);
55
56                 Topology topology = builder.build();
57                 log.info("\n\n{}", topology.describe());
58
59                 return topology;
60         }
61
62         public void start()
63         {
64                 log.info("Starting Stream-Processor");
65                 streams.start();
66         }
67
68         public void stop()
69         {
70                 log.info("Stopping Stream-Processor");
71                 streams.close();
72         }
73 }