背景
电力行业客户业务需求将网闸隔离内的Oracle数据实时同步到网闸外的集团统一数据湖平台Paimon内。通过LogMiner解析Redo日志捕获变更数据写入到Kafka之后，使用Flink消费Kafka写入到Paimon。Flink任务采用无状态(不开启checkpoint)、幂等写实现，数据流向如图所示：
客户反馈Flink任务提交遇到类加载相关的异常，详细日志如下所示：
Caused by: java.lang.LinkageError: loader constraint violation: when resolving interface method"org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer.committedOffsets(Lorg/apache/kafka/clients/consumer/OffsetResetStrategy;)Lorg/apache/flink/connector/kafka/source/enumerator/initializer/OffsetsInitializer;"the class loader (instance of org/apache/flink/util/ChildFirstClassLoader) of the current class,and the class loader (instance of sun/misc/Launcher$AppClassLoader) for the method's defining class,org/apache/flink/connector/kafka/source/enumerator/initializer/OffsetsInitializer,have different Class objects for the type org/apache/kafka/clients/consumer/OffsetResetStrategy used in the signature
本文通过深入剖析JDK类加载器及双亲委派模型、自定义类加载器及如何打破双亲委派模型、Flink定制类加载器机制等内容，由浅入深循序渐进地从原理层面弄明白并解决上述异常。文章发布于微信公众号:大数据从业者，其它均为转载，原创不易，欢迎您点赞关注推荐转发，谢谢！
Java是编译型面向对象编程语言，万事万物皆为类。Java源码类需要编译为二进制字节码.class文件才能加载使用，而类的加载就是JVM利用类加载器(ClassLoader)读取上述.class文件构造出类的定义。JDK已有类加载器包括BootStrap ClassLoader(启动类加载器、c/c++实现、加载核心库如rt.jar等)、Extension ClassLoader(扩展类加载器、Java实现、加载jre/lib/ext/*jar等)、App ClassLoader(应用/系统类加载器、Java实现、加载classpath路径*jar等)，三类加载器属于父子关系，如图所示：
如上图所示，前者都是后者的父类加载器。而所谓的双亲委派模型就是当JVM需要加载指定类时，父类加载器优先加载类，只有当父类加载器加载不到指定类时，子类加载器再加载指定类，如果一直加载不到，会提示ClassNotFound异常。强调一点：类加载器的父子关系只是逻辑概念，并非Java类的继承关系。实际上存在父子关系的ExtClassLoader和AppClassLoader都继承自URLClassLoader，如图所示：
通过如下的简单示例可以看出当前MyClass类的类加载器为AppClassLoader，并且AppClassLoader的父加载器就是ExtClassLoader，如图所示：
而为啥没打印出BootStrapClassLoader呢？正如上文所述它是c/c++实现，所以这里不会输出。至于双亲委派模型的实际效果演示以及如何打破、为什么打破这种模型，通过下面的自定义类加载器的实践更为直观。
自定义类加载器
双亲委派模型的弊端很多，比如父加载器无法访问子加载器加载的子类、无法支持同一类名不同版本的并行加载等等。为了打破双亲委派模型，可以实现自定义类加载器。上文知道ExtClassLoader和AppClassLoader都继承自URLClassLoader，最终继承自ClassLoader。本章节通过继承ClassLoader实现自己类加载器MyFileClassLoader，该加载器默认以AppClassLoader为父加载器，可以通过构造方法修改父加载器为null，打破双亲委派模型，如图所示：
通过实现ClassLoader. findClass方法，从指定路径通过类的全限定名加载指定类的二进制字节码.class文件，最后通过defineClass方法返回该类的定义，如图所示：
接下来，通过具体示例演示下效果：
1.与MyFileClassLoader同目录创建MyClass工具类，构造方法打印类唯一标识如图：
2.随便其他目录，如c:/MyClass目录，创建上述全限定类名相同的MyClass类，构造方法打印类唯一标识1. Constructor，通过javac编译生成.class文件，如图所示：
3.创建MyFileClassLoader测试类MyFileClassLoaderDemo，具体代码如图所示：
4.方式1为默认父加载器AppClassLoader，即遵循双亲委派模型，类加载器MyFileClassLoader优先让AppClassLoader加载。因为AppClassLoader加载classpath下jar，加载到了idea中的MyClass类，输出2. Constructor日志。所以，MyFileClassLoader不再加载类，执行结果如图所示：
5.方式2为重置父加载器为null，即打破双亲委派模型，类加载器MyFileClassLoader直接加载指定类，加载到指定的c:/MyClass目录类，输出1. Constructor日志，执行结果如图所示：
完整示例见github：
https://github.com/felixzh2020/felixzh-java/tree/main/MyClassLoader
实际上，稍微复杂点的项目都会使用自定义类加载器调整指定类的加载顺序，实现各种类的动态加载，Flink中自己实现的类加载器见如下章节。
Flink类加载器
Flink任务Jar类别分为：第一类为classpath相关jar和flink/lib目录jar，通过AppClassLoader加载；第二类为flink/plugin目录jar，通过PluginClassLoader加载；第三类为用户代码jar，通过FlinkUserCodeClassLoader加载。这三类jar类加载器结构关系如图所示：
Java classpath就不多说了，flink/lib放置核心jar和公共连接器jar，flink/plugin放置插件jar(动态加载，不启用，不加载)，而用户代码jar是用户指定提交的业务代码jar及其依赖jar。FlinkUserCodeClassLoader分为两种：ParentFirstClassLoader、ChildFirstClassLoader。
当用户代码jar及其依赖jar与另外两类jar冲突不兼容时候，可以通过指定上述两种FlinkUserCodeClassLoader调整类加载器顺序解决，对应的配置参数为classloader.resolve-order。child-first表示优先从用户代码Jar加载，也就是打破了Java双亲委派模型的类加载方式。而parent-first则表示AppClassLoader 将优先于 FlinkUserCodeClassLoader 进行类加载，即遵循双亲委派模型。
除了调整用户代码jar加载方式，还有一个参数yarn.classpath.include-user-jar可以决定YARN模式下用户代码Jar是否添加或者如何添加到task manager进程的classpath。
为了避免全局classpath污染导致各种类冲突或者方法不存在等问题，个人推荐优先使用classloader.resolve-order=child-first、yarn.classpath.include-user-jar=DISABLED组合模式。
解决方法
回到本文的异常，根本原因是ChildFirstClassLoader和AppClassLoader分别加载了不同版本的org.apache.kafka.clients.consumer.OffsetResetStrategy类。当Flink用户代码的OffsetsInitializer.committedOffsets方法尝试使用这个类时，JVM检测到两个类定义不一致，违反了类加载约束。而对应的代码就是用户指定的Flink消费Kafka是offset策略部分，如图所示：
如图期望效果就是从已提交到Kafka的Offset开始消费，如果已提交的Offset不存在，则使用EARLIEST Offset作为重置策略！
解决方法1：继续保持classloader.resolve-order=child-first，就是用户jar继续由ChildFirstClassLoader加载；而设置yarn.classpath.include-user-jar=DISABLE，就是不添加用户jar到taskmanager classpath，也就是AppClassLoader加载不到用户jar！
解决方法2：修改classloader.resolve-order=parent-first，就是所有类都通过AppClassLoader加载，同一个加载器同一个类只会加载一次！
总结
本文通过深入剖析JDK类加载器及双亲委派模型、自定义类加载器及如何打破双亲委派模型、Flink定制类加载器机制等内容，由浅入深循序渐进地从原理层面弄明白并解决上述异常。文章发布于微信公众号:大数据从业者，其它均为转载，原创不易，欢迎您点赞关注推荐转发，谢谢！
