Spring Cloud StreamListener@Output KStream Serdes似乎不起作用

Question

我有一个流监听器，作为

@StreamListener(target = "requesti")
@SendTo("responseo")
public KStream<UUID,Account> process(KStream<UUID, Account> events) {
    // Predicate<UUID, Event> isAccount = (key, value) ->
    // value.getEntity().getClass().equals(Account.class);

    // @formatter:off
    return events
            //.filter(isAccount)
            .peek((key, value) -> {
                log.debug("Processing {} {}", key, value);
            });
            /*
            .filter(isAccount)
            .map((key, value) -> process(value))

            .peek((key, value) -> {
                log.debug("Processed {} {}", key, value);
            });
            */
    // @formatter:on

}

其中@input（“requesti”）配置如下所示；

spring.cloud.stream.kafka.streams.bindings.requesti.consumer.application-id=repo-event-consumer
spring.cloud.stream.bindings.requesti.destination=request
spring.cloud.stream.bindings.requesti.content-type=application/json
spring.cloud.stream.bindings.requesti.consumer.header-mode=raw

spring.cloud.stream.kafka.streams.bindings.responseo.consumer.application-id=repo-response-producer
spring.cloud.stream.bindings.responseo.destination=response
spring.cloud.stream.bindings.responseo.content-type=application/json
spring.cloud.stream.bindings.responseo.producer.header-mode=raw
spring.cloud.stream.bindings.responseo.producer.use-native-encoding=true
spring.cloud.stream.kafka.streams.bindings.responseo.producer.key-serde=org.springframework.kafka.support.serializer.JsonSerde
spring.cloud.stream.kafka.streams.bindings.responseo.producer.value-serde=org.springframework.kafka.support.serializer.JsonSerde

发送记录ProducerRecord(TOPIC=Request，Partition=null,headers=recordheaders(headers=[recordheader(key=Key_TypeId，值=[106,97,118,97,46,117,116,105,108,85,85,85,73,68])，readheader（键=TypeId,值=[117,107,46,111,114,103,46,99,97,112,117,108,116,101,115,46,99,117,98,101,118,105,99,101,115,97,99,111,117,110,116,109,100,101,108,46,65,99,117,117,117,117,100,101,108,46,65,99,117,117,116])，isReadOnly=true）,key=6f0f50e2-3add-4d22-a370-CAC66D016AF0,value=account()，回调org.springframework.kafka.core.kafkatemplate$$lambda$582/533392019@85AB964到主题请求分区2

默认的serdeConfig为

    spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.springframework.kafka.support.serializer.JsonSerde
spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.springframework.kafka.support.serializer.JsonSerde

回购

Answer 1

下面的示例演示了jsonserde如何使用Kafka Streams绑定器进行出站：https://github.com/schacko-samples/json-serde-example。运行示例并确保其工作。查看application.yml以获得配置细节。我在提供的自述文件中添加了一些细节。