Spring kafka 2.2 type mappings class loader mismatch

[idkw]

Hi,

Following my question on StackOverflow which revealed to be actually a bug :
https://stackoverflow.com/questions/53559944/spring-kafka-2-2-type-mappings-class-loader-mismatch

Here it is again below for reference :

I'm trying to use the new type mappings feature introduced in Spring Kafka 2.2 :

Scroll down to "Mapping Types" :
https://docs.spring.io/spring-kafka/reference/htmlsingle/#serdes

On the producer side, I registered a mapping as follow :

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

And on the consumer as follow :

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

However, when sending an event to Kafka on the producer side of the com.myfoo.Foo class, the classId header that is added to the record is com.myfoo.Foo instead of foo. Thus on the consumer side, it fails to deserialize because com.myfoo.Foo is unknown.

I have narrowed the issue down to this method in 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)));
	}
}

When debugging through the execution of the serialization of a kafka record, the execution actually goes to the else branch, whereas in my understanding it should actually go to the if branch and add foo as a header. Instead it adds com.myfoo.Foo to the header.

The culprit seems to be related to a class loader mismatch but I'm not sure if it's actually a bug or something stupid I did on my end.

Basically, the classIdMapping map is correctly filled with com.myfoo.Foo as a key and the UTF-8 byte[] representation of foo as the corresponding value.

However during the if chec, the clazz parameter is a Class loaded by a different class loader as what was stored in the classIdMapping map, thus the hashcodes are different and it goes to the else branch.

Thanks

[garyrussell]

While this is easy enough to fix on the producer side by using the class name instead of the Class<?> itself; I don't see a solution on the consumer side (aside from the application injecting the deserializer into the consumer factory). If the deserializer is loaded by a different classloader, the application will get a ClassCastException when trying to access the value of the ConsumerRecord. What am I missing?

[artembilan]

That's good observation, @garyrussell , indeed.

I see that ConfigDef does this:

case CLASS:
                    if (value instanceof Class)
                        return value;
                    else if (value instanceof String)
                        return Class.forName(trimmed, true, Utils.getContextOrKafkaClassLoader());
                    else
                        throw new ConfigException(name, value, "Expected a Class instance or class name.");

Where that Utils.getContextOrKafkaClassLoader() is this:

public static ClassLoader getContextOrKafkaClassLoader() {
        ClassLoader cl = Thread.currentThread().getContextClassLoader();
        if (cl == null)
            return getKafkaClassLoader();
        else
            return cl;
}

I wonder if we can trick that Thread.currentThread().setContextClassLoader(beanFactoryClassLoader) somehow during KafkaConsumer instantiation in the DefaultKafkaConsumerFactory...

[garyrussell]

That should work, but I'd like to understand why the kafka class loader is different (Utils.getClassLoader()). Something else must be loading Utils before the application context is created and started.

@idkw Can you provide a small example that exhibits this behavior and/or the output of -verbose JVM arg together with a DEBUG log? So we can figure out what's going on with the class loaders?

[artembilan]

However I agree that it doesn't matter: when we call DefaultKafkaConsumerFactory.createKafkaConsumer() it is done in the application context thread, therefore its ClassLoader should be in use.

The simple project to reproduce and play would be really helpful.

Thanks

[idkw]

Thanks for the analysis.
I tried to reproduce the issue with a minimalistic project but could achieve the same condition. The mapping works properly here.
I'll try adding more configuration inspired from the bigger project with the issue to reproduce (I can't share the bigger project because it is closed source).

[garyrussell]

Thanks for trying; running with -verbose (and DEBUG log) might help track down what is loading the Utils class in a different class loader.

[idkw]

After some digging, I may have find some clues.
In my actual project, we are using several internal libraries and also spring boot dev tools.
I've stumbled onto this documentation warning :
https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#using-boot-devtools-customizing-classload

Apparently, this might cause our issue.
When I launch the app via my IDE (Intellij) it effectively loads some classes with LauncherAppClassLoader and others with RestartClassLoader which then causes these class loader mismatches.

When I launch the app from the built spring-boot executable jar no class cast exception occurs.

The documentation also says :

Restart functionality does not work well with objects that are deserialized by using a standard ObjectInputStream. If you need to deserialize data, you may need to use Spring’s ConfigurableObjectInputStream in combination with Thread.currentThread().getContextClassLoader().

Unfortunately, several third-party libraries deserialize without considering the context classloader. If you find such a problem, you need to request a fix with the original authors.

So I wonder if I have to add dev tools class cloader exceptions or if it is something to fix on spring-kafka side ?

[artembilan]

Sounds like there is nothing to fix on Spring for Apache Kafka side. Plus it looks like Kafka client really uses that Thread.currentThread().getContextClassLoader() to load classes from the provided properties.

So, that's unfortunate for you that you have bumped to this class loader issue.

Anyway would be great to have a simple Spring Boot app with those devtools to play with on our side. We really may find something we can reconsider.

Thanks

[idkw]

Hi,
I've finally managed to reproduce the issue on a minimalistic project.
This is definitively an issue with the way spring-boot-devtools uses two class loader to implement the hot restart feature.
When disabling the restart feature or removing the spring-boot-devtools dependency everything works fine.
The issue only happens when working with the dev tools in the IDE.

Here is the example project with the steps to reproduce in the included README
https://github.com/idkw/spring-kafka-894

Thanks

[artembilan]

When the envent is sent into kafka, if you debug the class org.springframework.kafka.support.converter.AbstractJavaTypeMapper you should see that on line 142 the clazz parameter is of class com.example.idkw.issue894.ProducerDto with class loader RestartClassLoader whereas what was stored in the classIdMapping map is the same class but with the class loader Launcher$AppClassLoader

So, that sounds like our original idea to store an FQCN as a string instead of Class<?> object in the classIdMapping should help here. But it still not clear why AbstractJavaTypeMapper initialization is loaded not by the RestartClassLoader...

Will investigate later.

[artembilan]

OK. I found the problem. The JsonSerializer has this code:

	mappingsMap.put(split[0].trim(),
						ClassUtils.forName(split[1].trim(), JsonSerializer.class.getClassLoader()));

Meanwhile Spring Boot Dev tool recommends to use a Thread.currentThread().getContextClassLoader() first.

[artembilan]

About your concern, Gary, during deserialization.
The AbstractJavaTypeMapper uses a ClassUtils.getDefaultClassLoader() which is supplied with the Thread.currentThread().getContextClassLoader().
So, that one should work as well.