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.*;
7 import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
8 import org.apache.kafka.streams.state.QueryableStoreTypes;
9 import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
10 import org.apache.kafka.streams.state.WindowBytesStoreSupplier;
11 import org.springframework.kafka.support.serializer.JsonSerde;
12 import org.springframework.kafka.support.serializer.JsonSerializer;
14 import java.time.Duration;
15 import java.time.ZoneId;
16 import java.time.ZonedDateTime;
18 import java.util.Properties;
19 import java.util.stream.Collectors;
23 public class PopularStreamProcessor
25 public static final String KEY_VALUE_STORE_NAME = "popular";
26 public static final String WINDOW_STORE_NAME = "popular-windows";
27 public static final Duration WINDOW_SIZE = Duration.ofSeconds(30);
30 public final KafkaStreams streams;
33 public PopularStreamProcessor(
36 Properties properties,
38 WindowBytesStoreSupplier windowBytesStoreSupplier,
39 KeyValueBytesStoreSupplier keyValueBytesStoreSupplier)
41 Topology topology = PopularStreamProcessor.buildTopology(
45 windowBytesStoreSupplier,
46 keyValueBytesStoreSupplier);
48 streams = new KafkaStreams(topology, properties);
51 static Topology buildTopology(
55 WindowBytesStoreSupplier windowBytesStoreSupplier,
56 KeyValueBytesStoreSupplier keyValueBytesStoreSupplier)
58 StreamsBuilder builder = new StreamsBuilder();
61 .stream(inputTopic, Consumed.with(inKeySerde(), inValueSerde()))
62 .peek((user, userWord) -> log.info("{}: {} -> {}", inputTopic, user, userWord))
63 .map((key, userWord) -> new KeyValue<>(Word.of(userWord.getWord()), Word.of(userWord.getWord())))
64 .peek((key, value) -> log.info("mapped: {} -> {}", key, value))
66 .windowedBy(TimeWindows.ofSizeWithNoGrace(WINDOW_SIZE))
69 .<Word, Long>as(windowBytesStoreSupplier)
70 .withKeySerde(new JsonSerde<>(Word.class).noTypeInfo())
71 .withValueSerde(Serdes.Long()))
72 .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
74 .peek((windowedWord, counter) -> log.info("windowed: {} -> {}", windowedWord, counter))
75 .map((windowedWord, counter) -> new KeyValue<>(
77 ZonedDateTime.ofInstant(windowedWord.window().startTime(), zone),
78 ZonedDateTime.ofInstant(windowedWord.window().endTime(), zone),
79 windowedWord.key().getWord()),
80 WordCounter.of(windowedWord.key().getWord(), counter)))
81 .peek((windowedWord, wordCounter) -> log.info("results: {} -> {}", windowedWord, wordCounter))
84 .<WindowedWord, WordCounter>as(keyValueBytesStoreSupplier)
85 .withKeySerde(new JsonSerde<>(WindowedWord.class).noTypeInfo())
86 .withValueSerde(new JsonSerde<>(WordCounter.class).noTypeInfo()))
88 .peek((windowedWord, wordCounter) -> log.info("output: {} -> {}", windowedWord, wordCounter))
89 .to(outputTopic, Produced.with(outKeySerde(), outValueSerde()));
91 Topology topology = builder.build();
92 log.info("\n\n{}", topology.describe());
97 ReadOnlyKeyValueStore<WindowedWord, WordCounter> getStore()
99 return streams.store(StoreQueryParameters.fromNameAndType(KEY_VALUE_STORE_NAME, QueryableStoreTypes.keyValueStore()));
104 log.info("Starting Stream-Processor");
110 log.info("Stopping Stream-Processor");
116 public static JsonSerde<User> inKeySerde()
118 return new JsonSerde<>(User.class);
121 public static JsonSerde<Word> inValueSerde()
123 return new JsonSerde<>(Word.class);
126 public static JsonSerde<WindowedWord> outKeySerde()
131 public static JsonSerde<WordCounter> outValueSerde()
136 public static <T> JsonSerde<T> serde(boolean isKey)
138 JsonSerde<T> serde = new JsonSerde<>();
140 Map.of(JsonSerializer.TYPE_MAPPINGS, typeMappingsConfig()),
145 private static String typeMappingsConfig()
147 return typeMappingsConfig(WindowedWord.class, WordCounter.class);
150 public static String typeMappingsConfig(Class wordClass, Class wordCounterClass)
154 "counter", wordCounterClass)
157 .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
158 .collect(Collectors.joining(","));