防止在Spring Cloud Stream中使用\uu TypeId\uuuu

Question

我们有一个流氓生产者将Kafka Header__TypeId__设置为一个类，该类是生产者的一部分，但不是使用Kafka Streams绑定器在Spring Cloud Stream应用程序中实现的消费者的一部分。它导致了一个异常java.lang.IllegalArgumentException：类“com.bad.MyClass”不在受信任的包中：[java.util、java.lang、de.datev.pws.loon.dcp.foreignmodels.*]。如果您认为这个类可以安全地反序列化，请提供它的名称。如果序列化仅由受信任的源完成，您也可以启用信任所有 (*).

我们如何确保在消费者内部忽略这个TypeId头？

一些stackoverflow答案指向spring。json。使用类型headers=false，但它似乎是一个“旧”属性，不再有效。

应用yaml：

spring:
  json.use.type.headers: false
  application:
    name: dcp-all
  kafka:
    bootstrap-servers: 'xxxxx.kafka.dev.dvint.de:9093'
  cloud:
    stream:
      kafka:
        streams:
          binder:
            required-acks: -1 # all in-sync-replicas

...

堆栈跟踪：

    at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.getClassIdType(DefaultJackson2JavaTypeMapper.java:129)
    at org.springframework.kafka.support.mapping.DefaultJackson2JavaTypeMapper.toJavaType(DefaultJackson2JavaTypeMapper.java:103)
    at org.springframework.kafka.support.serializer.JsonDeserializer.deserialize(JsonDeserializer.java:569)
    at org.apache.kafka.streams.processor.internals.SourceNode.deserializeValue(SourceNode.java:58)
    at org.apache.kafka.streams.processor.internals.RecordDeserializer.deserialize(RecordDeserializer.java:66)
    at org.apache.kafka.streams.processor.internals.RecordQueue.updateHead(RecordQueue.java:176)
    at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:112)
    at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:304)
    at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:960)
    at org.apache.kafka.streams.processor.internals.TaskManager.addRecordsToTasks(TaskManager.java:1068)
    at org.apache.kafka.streams.processor.internals.StreamThread.pollPhase(StreamThread.java:962)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:751)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:604)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:576)

这是一个单元测试

    @Test
    void consumeWorksEvenWithBadTypesHeader() throws JsonProcessingException, InterruptedException {
        Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka);
        producerProps.put("key.serializer", StringSerializer.class.getName());
        DefaultKafkaProducerFactory<String, String> pf = new DefaultKafkaProducerFactory<>(producerProps);

        List<Header> headers = Arrays.asList(new RecordHeader("__TypeId__", "com.bad.MyClass".getBytes()));
        ProducerRecord<String,String> p = new ProducerRecord(TOPIC1, 0, "any-key",
            "{ ... some valid JSON ...}", headers);

        try {
            KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
            template.send(p);

            ConsumerRecord<String, String> consumerRecord = KafkaTestUtils.getSingleRecord(consumer, TOPIC2, DEFAULT_CONSUMER_POLL_TIME);

            // Assertions ...
        } finally {
            pf.destroy();
        }
    }




Answer 1

您有两种选择：

1. 在producer端，将属性设置为省略添加类型信息头
2. 在使用者端，将属性设置为不使用类型信息标头

https://docs.spring.io/spring-kafka/docs/current/reference/html/#json-塞尔德

它不是一个“旧”的财产。

    /**
     * Kafka config property for using type headers (default true).
     * @since 2.2.3
     */
    public static final String USE_TYPE_INFO_HEADERS = "spring.json.use.type.headers";

它需要在消费者属性中设置。