如何为Kafka KeyValueStore设置值解/序列化程序？

Question

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

@StreamListener public void process(@input（“input”）ktable myMessages){

我在application.yml中配置了使用者，如下所示：

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

我缺少什么配置，以便使用我的自定义值解/序列化程序？解/串行化程序是否需要在特定的包中才能找到？

Answer 1

在application.yml中使用了错误的配置键。而不是key-deserializer：应该是keyserde：并且不是value-deserializer：应该是valueSerde。下面是正确的配置：

spring.cloud.stream.kafka.streams.bindings.input：consumeralizedas：all-messages keyserde：org.apache.kafka.common.serialization.serdes$stringserde valueserde：com.me.mymessageserde生产者：keyserde：org.apache.kafka.common.serialization.serdes$stringserde valueserde：com.me.mymessageserde