Spring Cloud Stream Elmhrust.RELEASE 无法更改 Serde

Question

我无法使用文档 (https://docs.spring.io/spring-cloud-stream/docs/Elmhurst.RELEASE/reference/htmlsingle/#_configuration_options_3) 中指定的语法更改通道（或绑定）的 Serde。

假设我的频道是pcin，我知道我应该使用以下属性spring.cloud.stream.kafka.streams.bindings.pcin.producer.valueSerde 和spring.cloud.stream.kafka.streams.bindings.pcin.producer.keySerde 指定valueSerde 和keySerde。

但是，我收到一个异常：

Caused by: org.apache.kafka.streams.errors.StreamsException: A serializer (key: org.apache.kafka.common.serialization.StringSerializer / value: org.apache.kafka.common.serialization.StringSerializer) is not compatible to the actual key or value type (key type: java.lang.String / value type: java.lang.Long). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.

我正在尝试改编 Josh Long 的 Spring Tips 中的示例：https://github.com/spring-tips/spring-cloud-stream-kafka-streams

我刚刚将PageViewEventProcessor类改成如下：

@Component
        public static class PageViewEventProcessor {

                @StreamListener
                @SendTo(AnalyticsBinding.PAGE_COUNT_OUT)
                public KStream<String, Long> process(@Input(AnalyticsBinding.PAGE_VIEWS_IN) KStream<String, PageViewEvent> events) {
                        return events
                            .filter((key, value) -> value.getDuration() > 10)
                            .map((key, value) -> new KeyValue<>(value.getPage(), value.getDuration()))
                            .groupByKey()
                            .aggregate(()-> 0L, 
                                    (cle, val, valAgregee) -> valAgregee + val, 
                                    Materialized.as(AnalyticsBinding.PAGE_COUNT_MV))

                            .toStream();
                }
        }

我不计算事件（页面访问）的数量，而是计算每次访问的持续时间总和。

这是 application.properties 的摘录（来自 Spring 提示示例）：

# page counts out
spring.cloud.stream.bindings.pcout.destination=pcs
spring.cloud.stream.bindings.pcout.producer.use-native-encoding=true
spring.cloud.stream.kafka.streams.bindings.pcout.producer.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.pcout.producer.value-serde=org.apache.kafka.common.serialization.Serdes$LongSerde
#
# page counts in
spring.cloud.stream.bindings.pcin.destination=pcs
spring.cloud.stream.bindings.pcin.consumer.use-native-decoding=true
spring.cloud.stream.bindings.pcin.group=pcs
spring.cloud.stream.bindings.pcin.content-type=application/json
spring.cloud.stream.bindings.pcin.consumer.header-mode=raw
spring.cloud.stream.kafka.streams.bindings.pcin.consumer.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.pcin.consumer.value-serde=org.apache.kafka.common.serialization.Serdes$LongSerde

还有其他必要的更改吗？

Answer 1

pcin 是否绑定在消费者（输入）上？如果是这样，您应该使用这些属性作为 spring.cloud.stream.kafka.streams.bindings.pcin.consumer.valueSerde and spring.cloud.stream.kafka.streams.bindings.pcin.consumer.keySerde

您的传入值类型是PageViewEvent。但是，您将值 Serde 设置为 LongSerde。

您可以完全删除此属性：spring.cloud.stream.bindings.pcin.consumer.use-native-decoding=true，并让框架为您进行 JSON 转换。这样，传入类型会自动转换为PageViewEvent，而无需您显式提供值 Serde。

如果您必须提供一个值 Serde（在这种情况下，本机解码属性必须设置为 true），那么您必须提供一个正确的 JsonSerde 作为值 Serde。

更新：

通过以下更改，我能够运行应用程序而不会出现任何错误。

我这样修改了你的代码。

@StreamListener
@SendTo(AnalyticsBinding.PAGE_COUNT_OUT)
public KStream<String, Long> process(@Input(AnalyticsBinding.PAGE_VIEWS_IN) KStream<String, PageViewEvent> events) {

                    return events
                            .filter((key, value) -> value.getDuration() > 10)
                            .map((key, value) -> new KeyValue<>(value.getPage(), value.getDuration()))
                            .groupByKey(Serialized.with(Serdes.String(), Serdes.Long()))
                            .aggregate(()-> 0L,
                                    (cle, val, valAgregee) -> valAgregee + val,
                                    Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as(AnalyticsBinding.PAGE_COUNT_MV)
                                    .withKeySerde(Serdes.String()).withValueSerde(Serdes.Long())
                            )
                            .toStream();
                }

groupByKey 和 aggregate 调用中的内部 Serdes 是必要的，因为它们不同于默认的键/值 Serde 组合。

我还更改了您的配置并清理了它：

#
# defaults
spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.mms=1000
spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
#
# page views out
spring.cloud.stream.bindings.pvout.destination=pvs
#
# page views in
spring.cloud.stream.bindings.pvin.destination=pvs
#
# page counts out
spring.cloud.stream.bindings.pcout.destination=pcs
spring.cloud.stream.bindings.pcout.producer.use-native-encoding=true
spring.cloud.stream.kafka.streams.bindings.pcout.producer.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.pcout.producer.value-serde=org.apache.kafka.common.serialization.Serdes$LongSerde
#
# page counts in
spring.cloud.stream.bindings.pcin.destination=pcs
spring.cloud.stream.bindings.pcin.consumer.use-native-decoding=true
spring.cloud.stream.bindings.pcin.group=pcs
spring.cloud.stream.kafka.streams.bindings.pcin.consumer.key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.cloud.stream.kafka.streams.bindings.pcin.consumer.value-serde=org.apache.kafka.common.serialization.Serdes$LongSerde