在 Kafka JSON 序列化中发送超类型作为类型信息

Question

我正在使用 Spring Boot 使用 Kafka 将数据从一个应用程序发送到另一个应用程序。

我的设计使用接口来声明正在发送的数据：

package domain;

interface Data {
    public String getData();
    public void setData(String data);
}

制片人

在源应用程序中，我将此接口实现为 db 实体。

package persistence;

@Data
class DataEntity implements Data {
    private String data;  // lombok generates getter/setters
}

添加实体后，我想使用 KafkaTemplate 将其作为更新发送到 Kafka

@Component
class DataPublisher implements ApplicationListener<DataEvent> {
    @Autowired private KafkaTemplate<String,Data> template;

    // I left out DataEvent which is a straightforward ApplicationEvent override
    @EventListener(classes = DataEvent.class)
    public void onApplicationEvent(DataEvent event) {
        template.send("data", (Data) event.getSource());
    }
}
// triggered by this call in a service
    eventPublisher.publishEvent(new DataEvent(updatedData));

序列化是通过属性完成的

spring:
    kafka:
        consumer:
            value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
            properties.spring.json.value.default.type: domain.Data

查看kafkacat输出，数据发送正常。

消费者

在接收端我有

@KafkaListener(topics = "data")
public void dataUpdated(@Payload Data data) {
    dataService.updateData(data);
}

导致

原因：java.lang.IllegalArgumentException：类“persistence.DataEntity”不在受信任的包中[...]

我完全理解——序列化程序发送一个persistence.DataEntity 对象，但客户端需要一个domain.Data 对象。但这就是设计的本意；我希望客户只知道域包，而不是它的persistence 实现。 （作为一个附带问题，我在哪里可以看到这种类型的标头？它不在编码的 json AFAICT 中，我错过了什么？）

所以问题是：如何强制 Spring JsonDeserializer 发送 domain.Data 作为序列化数据类型？

我确实在序列化程序类中找到了一个 TYPE_MAPPING 属性，但它唯一的文档是它“将 [s] 类型映射添加到类型映射器：'foo:com.Foo,bar:com.Bar'”没有解释任何东西，我找不到示例用法。

编辑：

我确实添加了

spring.kafka.producer.properties.spring.json.type.mapping=domain.Data:persistence.DataEntity

到生产者的属性，但这没有帮助。

Answer 1

见the documentation。

您必须提供双方的映射。

但是，您应该使用BytesDeserializer 和BytesJsonMessageConverter，而不是使用JsonDeserializer（只需添加一个@Bean，Boot 就会将其连接到容器工厂）。

这样框架会自动转换成参数类型。

再次，请参阅the documentation。