StreamsBuilder builder = new StreamsBuilder();
builder
- .stream(
- inputTopic,
- Consumed.with(
- new JsonSerde<>(User.class),
- new JsonSerde<>(Word.class)))
+ .stream(inputTopic, Consumed.with(inKeySerde(), inValueSerde()))
.map((key, word) -> new KeyValue<>(word, word))
.groupByKey()
.count(
Materialized
.<Word, Long>as(storeSupplier)
- .withKeySerde(new JsonSerde<>().copyWithType(Word.class).forKeys()))
+ .withKeySerde(new JsonSerde<>(Word.class))) // No headers are present: fixed typing is needed!
.toStream()
.map((word, counter) -> new KeyValue<>(word, WordCounter.of(word, counter)))
.to(outputTopic, Produced.with(outKeySerde(), outValueSerde()));
+ public static JsonSerde<User> inKeySerde()
+ {
+ return new JsonSerde<>(User.class);
+ }
+
+ public static JsonSerde<Word> inValueSerde()
+ {
+ return new JsonSerde<>(Word.class);
+ }
+
public static JsonSerde<Word> outKeySerde()
{
return serde(true);
private static String typeMappingsConfig()
{
- return typeMappings()
- .entrySet()
- .stream()
- .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
- .collect(Collectors.joining(","));
+ return typeMappingsConfig(Word.class, WordCounter.class);
}
- private static Map<String, Class> typeMappings()
+ public static String typeMappingsConfig(Class wordClass, Class wordCounterClass)
{
return Map.of(
- "word", Word.class,
- "counter", WordCounter.class);
+ "word", wordClass,
+ "counter", wordCounterClass)
+ .entrySet()
+ .stream()
+ .map(entry -> entry.getKey() + ":" + entry.getValue().getName())
+ .collect(Collectors.joining(","));
}
}