- 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))
- );
+ KTable<String, User> users = builder
+ .stream(usersInputTopic)
+ .toTable(Materialized.with(null, new JsonSerde().copyWithType(User.class)));