.withValueSerde(new JsonSerde().copyWithType(User.class)));
KStream<String, Ranking> rankings = builder
.<Key, Ranking>stream(rankingInputTopic)
.withValueSerde(new JsonSerde().copyWithType(User.class)));
KStream<String, Ranking> rankings = builder
.<Key, Ranking>stream(rankingInputTopic)