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 发送事件时，添加到记录中的 classId 标头是 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 映射正确地填充了 com.myfoo.Foo 作为键和 foo 的 UTF-8 byte[] 表示作为对应的值。

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

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

谢谢

Answer 1

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

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

作为一种解决方法，同时我们修复了该问题，我们建议使用 DefaultKafkaProducerFactory 配置及其 setValueSerializer() 来填充 JsonSerializer bean 引用。这样，您需要通过其setTypeMapper() 填充类映射，因此，DefaultJackson2JavaTypeMapper 上的setIdClassMapping()。