- KTable<String, String> users = builder.table(usersInputTopic);
- KStream<String, String> rankings = builder.stream(rankingInputTopic);
+ JsonSerde valueSerde = new JsonSerde();
+ valueSerde.configure(Map.of(
+ JsonDeserializer.TYPE_MAPPINGS,
+ "user:" + Key.class.getName() + "," +
+ "ranking:" + Ranking.class.getName() + "," +
+ "userdata:" + User.class.getName() + "," +
+ "userranking:" + UserRanking.class.getName()
+ ), false);
+ KTable<String, User> users = builder.table(
+ usersInputTopic,
+ Consumed.with(null, valueSerde.copyWithType(User.class))
+ );
+ KStream<String, Ranking> rankings = builder.stream(rankingInputTopic);