Spring kafka 2.2类型映射类加载器不匹配

Question

我正在尝试使用 Spring Kafka 2.2 中引入的新类型映射功能:

向下滚动到“映射类型”:
https://docs.spring.io/spring-kafka/reference/htmlsingle/#serdes

在生产者方面，我注册了一个映射，如下所示:

senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
senderProps.put(JsonSerializer.TYPE_MAPPINGS, "foo:com.myfoo.Foo");

对于消费者如下:

consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
consumerProps.put(JsonDeSerializer.TYPE_MAPPINGS, "foo:com.yourfoo.Foo");

但是，当 com.myfoo.Foo 的生产者端向 Kafka 发送事件时class，添加到记录中的 classId header 为 com.myfoo.Foo而不是foo 。因此，在消费者方面，它无法反序列化，因为 com.myfoo.Foo未知。

我已将问题范围缩小到 spring-kafka 中的此方法:

https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/support/converter/AbstractJavaTypeMapper.java#L142

protected void addHeader(Headers headers, String headerName, Class<?> clazz) {
    if (this.classIdMapping.containsKey(clazz)) {
        headers.add(new RecordHeader(headerName, this.classIdMapping.get(clazz)));
    }
    else {
        headers.add(new RecordHeader(headerName, clazz.getName().getBytes(StandardCharsets.UTF_8)));
    }
}

当通过kafka记录的序列化执行进行调试时，执行实际上转到else分支，而根据我的理解，它实际上应该转到if分支并添加 foo作为标题。相反，它添加了 com.myfoo.Foo到标题。

罪魁祸首似乎与类加载器不匹配有关，但我不确定这实际上是一个错误还是我做了一些愚蠢的事情。

基本上，classIdMapping map 正确填充 com.myfoo.Foo作为键和 foo 的 UTF-8 byte[] 表示形式作为对应的值。

但是在 if 检查期间，clazz参数是由不同的类加载器加载的类，与 classIdMapping 中存储的类加载器不同。映射，因此哈希码不同，它转到 else 分支。

这实际上是 String-Kafka 方面的错误还是我配置错误？

谢谢

Answer 1

这看起来确实像是我们实现中的一个遗漏。请为 Apache Kafka 项目的 Spring 提出问题:https://github.com/spring-projects/spring-kafka 。我们的想法是不要使用Class<?>用于映射，但其完全限定类名。

看来您的应用程序是多类加载器，例如它是网络之一。 Apache Kafka 客户端加载 senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);在它的 ClassLoader 中，同时其他一切都由应用程序上下文 ClassLoader 完成。

作为解决方法，在解决问题的同时，我们建议使用 DefaultKafkaProducerFactory配置及其 setValueSerializer()填充 JsonSerializer bean 引用。这样，您需要通过其 setTypeMapper() 填充类映射。因此，setIdClassMapping()关于DefaultJackson2JavaTypeMapper .