带kafka streams绑定器的Spring cloud stream：如何为流处理器设置trusted.packages（与消费者和生产者不同）

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)
    }}
}

这些类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 (*).

Answer 1

您可以在...binder.configuration下设置任意配置属性（Kafka streams属性、serdes使用的属性等）：

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