- KStream<String, String> source = builder.stream(inputTopic);
- source
- .map((username, word) ->
- {
- try
- {
- String key = mapper.writeValueAsString(Key.of(username, word));
- return new KeyValue<>(key, word);
- }
- catch (JsonProcessingException e)
- {
- throw new RuntimeException(e);
- }
- })
+ builder
+ .stream(
+ inputTopic,
+ Consumed.with(
+ new JsonSerde<>(User.class),
+ new JsonSerde<>(Word.class)))
+ .map((key, word) -> new KeyValue<>(word, word))