java - 如何为 Kafka KeyValueStore 设置值解/序列化器？

Question

我将来自 Kafka 主题的消息存储在 KeyValueStore 中，以便稍后查询它们。我创建一个 KTable 如下:

@StreamListener
public void process(@Input("input") KTable<String,MyMessage> myMessages) {

我在 application.yml 中配置了消费者，如下所示:

更新的反/序列化程序包

spring.cloud.stream.kafka.streams.bindings.input:
consumer:
materializedAs: all-messages
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: com.me.MyMessageDeserializer
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: com.me.MyMessageSerializer

但是，当我从 KeyValueStore 读取时，键会作为字符串正确返回，但返回的值是字节数组而不是 MyMessage。由于某种原因，我的自定义解串器没有被使用。我尝试自己反序列化该消息，但我的反序列化器因异常而崩溃。我在序列化器上放置了一个断点，但它从未被调用。我很清楚，我的串行器或解串器都没有被使用。

我缺少什么配置才能使用我的自定义值解/序列化器？反序列化器是否需要位于特定的包中才能找到？

Answer 1

application.yml 中使用了错误的配置键。而不是 key-deserializer:它应该是 keySerde:而不是 value-deserializer:它应该是 valueSerde。下面是正确的配置:

spring.cloud.stream.kafka.streams.bindings.input:
消费者:
物化为:所有消息
keySerde:org.apache.kafka.common.serialization.Serdes$StringSerde
valueSerde:com.me.MyMessageSerde
制片人:
keySerde:org.apache.kafka.common.serialization.Serdes$StringSerde
valueSerde:com.me.MyMessageSerde