1 package de.juplo.kafka.wordcount.popular;
3 import lombok.extern.slf4j.Slf4j;
4 import org.apache.kafka.common.serialization.Serdes;
5 import org.apache.kafka.streams.*;
6 import org.apache.kafka.streams.kstream.Consumed;
7 import org.apache.kafka.streams.kstream.Materialized;
8 import org.apache.kafka.streams.kstream.Produced;
9 import org.apache.kafka.streams.kstream.TimeWindows;
10 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
11 import org.apache.kafka.streams.state.QueryableStoreTypes;
12 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
13 import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
14 import org.springframework.kafka.support.serializer.JsonSerde;
15 import org.springframework.kafka.support.serializer.JsonSerializer;
17 import java.time.Duration;
18 import java.time.ZoneId;
19 import java.time.ZonedDateTime;
21 import java.util.Properties;
22 import java.util.stream.Collectors;
26 public class PopularStreamProcessor
28 public static final String KEY_VALUE_STORE_NAME = "popular";
29 public static final String WINDOW_STORE_NAME = "popular-windows";
30 public static final Duration WINDOW_SIZE = Duration.ofSeconds(30);
33 public final KafkaStreams streams;
36 public PopularStreamProcessor(
39 Properties properties,
41 WindowBytesStoreSupplier windowBytesStoreSupplier,
42 KeyValueBytesStoreSupplier keyValueBytesStoreSupplier)
44 Topology topology = PopularStreamProcessor.buildTopology(
48 windowBytesStoreSupplier,
49 keyValueBytesStoreSupplier);
51 streams = new KafkaStreams(topology, properties);
54 static Topology buildTopology(
58 WindowBytesStoreSupplier windowBytesStoreSupplier,
59 KeyValueBytesStoreSupplier keyValueBytesStoreSupplier)
61 StreamsBuilder builder = new StreamsBuilder();
64 .stream(inputTopic, Consumed.with(inKeySerde(), inValueSerde()))
65 .peek((user, userWord) -> log.info("{}: {} -> {}", inputTopic, user, userWord))
66 .map((key, userWord) -> new KeyValue<>(Word.of(userWord.getWord()), Word.of(userWord.getWord())))
67 .peek((key, value) -> log.info("mapped: {} -> {}", key, value))
69 .windowedBy(TimeWindows.ofSizeWithNoGrace(WINDOW_SIZE))
72 .<Word, Long>as(windowBytesStoreSupplier)
73 .withKeySerde(new JsonSerde<>(Word.class).noTypeInfo())
74 .withValueSerde(Serdes.Long()))
76 .peek((windowedWord, counter) -> log.info("windowed: {} -> {}", windowedWord, counter))
77 .map((windowedWord, counter) -> new KeyValue<>(
79 ZonedDateTime.ofInstant(windowedWord.window().startTime(), zone),
80 ZonedDateTime.ofInstant(windowedWord.window().endTime(), zone),
81 windowedWord.key().getWord()),
82 WordCounter.of(windowedWord.key().getWord(), counter)))
83 .peek((windowedWord, wordCounter) -> log.info("results: {} -> {}", windowedWord, wordCounter))
86 .<WindowedWord, WordCounter>as(keyValueBytesStoreSupplier)
87 .withKeySerde(new JsonSerde<>(WindowedWord.class).noTypeInfo())
88 .withValueSerde(new JsonSerde<>(WordCounter.class).noTypeInfo()))
90 .peek((windowedWord, wordCounter) -> log.info("output: {} -> {}", windowedWord, wordCounter))
91 .to(outputTopic, Produced.with(outKeySerde(), outValueSerde()));
93 Topology topology = builder.build();
94 log.info("\n\n{}", topology.describe());
99 ReadOnlyKeyValueStore<WindowedWord, WordCounter> getStore()
101 return streams.store(StoreQueryParameters.fromNameAndType(KEY_VALUE_STORE_NAME, QueryableStoreTypes.keyValueStore()));
106 log.info("Starting Stream-Processor");
112 log.info("Stopping Stream-Processor");
118 public static JsonSerde<User> inKeySerde()
120 return new JsonSerde<>(User.class);
123 public static JsonSerde<Word> inValueSerde()
125 return new JsonSerde<>(Word.class);
128 public static JsonSerde<WindowedWord> outKeySerde()
133 public static JsonSerde<WordCounter> outValueSerde()
138 public static <T> JsonSerde<T> serde(boolean isKey)
140 JsonSerde<T> serde = new JsonSerde<>();
142 Map.of(JsonSerializer.TYPE_MAPPINGS, typeMappingsConfig()),
147 private static String typeMappingsConfig()
149 return typeMappingsConfig(WindowedWord.class, WordCounter.class);
152 public static String typeMappingsConfig(Class wordClass, Class wordCounterClass)
156 "counter", wordCounterClass)
159 .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
160 .collect(Collectors.joining(","));