From 913c2c4ec0a3584f6be27c8341898d17c4260501 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Wed, 12 Jun 2024 23:08:57 +0200 Subject: [PATCH] query: 2.0.0 - (GREEN) Adjusted the implementation to the new expectations * Messages from the incomming topic, that is written by the `users` service can be serialized, although no type-information is conveyed via the headers. --- .../kafka/wordcount/query/QueryApplicationConfiguration.java | 1 - .../de/juplo/kafka/wordcount/query/QueryStreamProcessor.java | 5 ++++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java index 2ece744..6c7844d 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java @@ -92,7 +92,6 @@ public class QueryApplicationConfiguration JsonDeserializer.TYPE_MAPPINGS, "user:" + Key.class.getName() + "," + "ranking:" + Ranking.class.getName() + "," + - "userdata:" + User.class.getName() + "," + "userranking:" + UserRanking.class.getName()); return props; diff --git a/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java b/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java index 4749264..dcb1234 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java @@ -5,6 +5,7 @@ import jakarta.annotation.PreDestroy; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.*; +import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Materialized; @@ -57,7 +58,9 @@ public class QueryStreamProcessor StreamsBuilder builder = new StreamsBuilder(); KTable users = builder - .stream(usersInputTopic) + .stream( + usersInputTopic, + Consumed.with(null, new JsonSerde().copyWithType(User.class))) .toTable( Materialized .as(userStoreSupplier) -- 2.20.1