From 2f5828ee2bbd662c3c81c76961d00b871468c8b9 Mon Sep 17 00:00:00 2001 From: Kai Moritz Date: Tue, 11 Jun 2024 20:36:58 +0200 Subject: [PATCH] query: 2.0.0 - Values are serialized as JSON -- works, but is very confusing * The default-type is specified as a consumption-parameter in the command, that reads the input topic into the `KTable` via ``Consumed.with(..)``. * The resulting code is confusing, because the ``Consumed``-parameter is used for both, the consumption of the input topic _and_ the consumption of stored values, if read from the state-store. * Because of this, one might only think of the consumption of the stored values from the state-store, when looking at the ``Consumed.with()``- statement, and argue, why the type-mappings have to be specified here. --- .../query/QueryApplicationConfiguration.java | 1 - .../wordcount/query/QueryStreamProcessor.java | 15 ++++++++++++++- 2 files changed, 14 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 7da1712..07b78e4 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryApplicationConfiguration.java @@ -80,7 +80,6 @@ public class QueryApplicationConfiguration props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName()); props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class.getName()); - props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, User.class.getName()); // << Does not work without this! props.put( JsonDeserializer.TYPE_MAPPINGS, "user:" + Key.class.getName() + "," + 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 cc65fce..0692652 100644 --- a/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java +++ b/src/main/java/de/juplo/kafka/wordcount/query/QueryStreamProcessor.java @@ -13,9 +13,11 @@ import org.apache.kafka.streams.state.HostInfo; import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; +import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerde; import java.net.URI; +import java.util.Map; import java.util.Optional; import java.util.Properties; @@ -53,7 +55,18 @@ public class QueryStreamProcessor { StreamsBuilder builder = new StreamsBuilder(); - KTable users = builder.table(usersInputTopic); + 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 users = builder.table( + usersInputTopic, + Consumed.with(null, valueSerde.copyWithType(User.class)) + ); KStream rankings = builder.stream(rankingInputTopic); rankings -- 2.20.1