[FLINK-34996] Custom Deserializer can't be instantiated when connector-kafka installed into Flink Libs

• Type: Bug
• Status: Open
• Priority: Minor
• Resolution: Unresolved
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Hugo Gu
• Reporter: Hugo Gu
• Created: 03/Apr/24 09:39
• Updated: 19/Apr/24 05:47

Description

The current implementation of the 
KafkaValueOnlyDeserializerWrapper Class instantiates Deserializer from the ClassLoader of the KafkaValueOnlyDeserializerWrapper itself as following figure shows.
 

 
In case of both following conditions are met:
1. The connector-kafka get installed into Libs of Flink (rather than in the User Jar)
2. The user jar defines a customized Deserializer for Kafka Record. 
 
The instantiation of the custom deserializer will fail due to NoClassFound exception because it is indeed not available in the system class loader. 
 
As following figure illustrates
 

 
It can be fixed by using either UserCodeClassLoader or the ClassLoader of current Thread.

Issue Links

• links to: GitHub Pull Request #89