From f929a47a66a2c14a291133aaa624ad9c5696bd53 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Sat, 15 Jun 2024 07:34:30 +0200 Subject: [PATCH] query: 2.0.0 - (GREEN) Explicitly specifed the missing serde-config --- .../juplo/kafka/wordcount/query/QueryStreamProcessor.java | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) 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 bf27e2d..3a1665f 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java @@ -5,10 +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; +import org.apache.kafka.streams.kstream.*; import org.apache.kafka.streams.state.HostInfo; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.QueryableStoreTypes; @@ -74,7 +71,8 @@ public class QueryStreamProcessor .join(users, (ranking, user) -> UserRanking.of( user.getFirstName(), user.getLastName(), - ranking.getEntries())) + ranking.getEntries()), + Joined.keySerde(Serdes.String())) .toTable( Materialized .as(rankingStoreSupplier) -- 2.20.1