kotlin - 带有kafka流绑定(bind)器: how to set `trusted.packages` for a Stream Processor (that's different than consumer and producer)的 Spring Cloud 流

Question

我有一个简单的流处理器 (不是消费者/生产者)看起来像这样(Kotlin)

@Bean
fun processFoo():Function<KStream<FooName, FooAddress>, KStream<FooName, FooAddressPlus>> {
    return Function { input-> input.map { key, value ->
        println("\nPAYLOAD KEY: ${key.name}\n");
        println("\nPAYLOAD value: ${value.address}\n");
        val output = FooAddressPlus()
        output.address = value.address
        output.name = value.name
        output.plus = "$value.name-$value.address"
        KeyValue(key, output)
    }}
}

这些类(class) FooName , FooAddress和 FooAddressPlus与处理器在同一个包中。
这是我的配置文件:

spring.cloud.stream.kafka.binder:
  brokers: localhost:9093

spring.cloud.stream.function.definition: processFoo

spring.cloud.stream.kafka.streams.binder.functions.processFoo.applicationId: foo-processor
spring.cloud.stream.bindings.processFoo-in-0:
  destination: foo.processor
spring.cloud.stream.bindings.processFoo-out-0:
  destination: foo.processor.out

spring.cloud.stream.kafka.streams.binder:
  deserializationExceptionHandler: logAndContinue
  configuration:
    default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
    commit.interval.ms: 1000

运行处理器时出现此错误:

The class '<here_comes_package>.FooAddress' is not in the trusted packages: [java.util, java.lang]. 
If you believe this class is safe to deserialize, please provide its name. 
If the serialization is only done by a trusted source, you can also enable trust all (*).

使用 时，为所有内容设置受信任包的最佳方法是什么？ Kafka Streams Binder 流处理器 ? (没有消费者/生产者，只有流处理器)
非常感谢!

Answer 1

您可以在 ...binder.configuration 下设置任意配置属性(Kafka 流属性、serdes 使用的属性等)。 :

spring:
  cloud:
    stream:
      kafka:
        streams:
          binder:
            configuration:
              spring.json.trusted.packages: '*'