如何在春云流反序列化时信任所有包？

Question

该消费者不需要受信任的包:

@Bean
    fun berichtStateStoreBuilder() = Consumer<GlobalKTable<String, BerichtEvent>> {}

这突然发生了：

@Bean
    fun berichtStateStoreBuilder() = Consumer<KStream<ByteArray, ByteArray>> {
        it
            .transform({ EventTypeAwareTransformer(EVENT_TYPE_MAPPING, objectMapper) })
            .mapValues { v -> v.payload as BerichtEvent }
            .groupByKey(Grouped.with(Serdes.StringSerde(), JsonSerde()))
            .aggregate(
                { BerichtAggregator() },
                { _, event, aggregator -> aggregator.add(event) },
                Named.`as`("aggregate"),
                Materialized.`as`<String, BerichtAggregator, KeyValueStore<Bytes, ByteArray>>(BerichtStore.NAME)
                    .withKeySerde(Serdes.String())
                    .withValueSerde(JsonSerde(BerichtAggregator::class.java))
            )

我已经尝试了以下方法，但它们不起作用，因为我只得到以下错误:

Caused by: java.lang.IllegalArgumentException: The class 'at.wrwks.smp.controlling.event.BerichtEvent' 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 (*).
    at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:126)
    at org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper.toJavaType(DefaultJackson2JavaTypeMapper.java:100)
    at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:504)
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:55)
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
    ... 8 more

< li>

    @Bean
    fun defaultKafkaHeaderMapper(objectMapper: ObjectMapper): DefaultKafkaHeaderMapper {
        val mapper = DefaultKafkaHeaderMapper(objectMapper, "event_type")

        val rawMappedHeaders = HashMap<String, Boolean>()
        rawMappedHeaders[BaseEvent.EVENT_TYPE_HEADER] = true
        mapper.setRawMappedHeaders(rawMappedHeaders)
        mapper.addTrustedPackages("*")
        return mapper
    }

spring.cloud.stream.kafka.streams.binder.header-mapper-bean-name: defaultKafkaHeaderMapper

spring.cloud.stream.kafka.streams.binder.configuration.spring.json.use.type.headers: false

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

Spring Cloud Stream版本：3.1.2，带有Kafka Streams活页夹。

使用自定义JSON serde解决方法：

            .groupByKey(Grouped.with(Serdes.StringSerde(), Serdes.serdeFrom(
                SimpleJsonSerializer(objectMapper), SimpleJsonDeserializer(objectMapper, BerichtEvent::class.java)
            )))

Answer 1

虽然这次我已经这样做了几十次，但我忘记将目标类传递给构造函数 JsonSerde（）。这是正确的：

.groupByKey(Grouped.with(Serdes.StringSerde(), JsonSerde(BerichtEvent::class.java)))

显然，当没有类被传递时，就不能将任何包添加到受信任的包中。通过传递类，Serde将使用目标传递所属的包进行配置。

Answer 2

我刚刚测试了它，它对我来说效果很好...

@SpringBootApplication
public class So67059860Application {

    public static void main(String[] args) {
        SpringApplication.run(So67059860Application.class, args);
    }

    @Bean
    public Consumer<KStream<String, Foo>> input() {
        return str -> str.foreach((k, v) -> System.out.println(v));
    }

}

public class Foo {

    private String bar;

    public String getBar() {
        return this.bar;
    }

    public void setBar(String bar) {
        this.bar = bar;
    }

    @Override
    public String toString() {
        return "Foo [bar=" + this.bar + "]";
    }

}

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

spring.cloud.stream.kafka.streams.binder.configuration.spring.json.trusted.packages=*
spring.cloud.stream.kafka.streams.binder.configuration.spring.json.value.default.type=com.example.demo.Foo
spring.application.name=so67059860

spring.cloud.function.definition=input

#logging.level.root=debug

Foo [bar=baz]

启动2.4.4，云2020.0.2（SCSt 3,1.2）。

在JsonSerde中设置断点。configure（）以查看正在使用的属性。