Merge branch '1.5_v3.8.1_bugfix_classloader' into '1.5_v3.8.1'

[flinksql][udf类加载器和tableEnv不是同一个][19234]



See merge request !95

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/Main.java

@@ -249,14 +249,14 @@ private static void registerUDF(SqlTree sqlTree, List<URL> jarURList, URLClassLo
                                     StreamTableEnvironment tableEnv)
             throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException {
         //register urf
-        // udf函数只能由appclassloader加载
-        ClassLoader superClassLoader = Thread.currentThread().getContextClassLoader().getParent();
+        // udf和tableEnv须由同一个类加载器加载
+        ClassLoader levelClassLoader = tableEnv.getClass().getClassLoader();
         URLClassLoader classLoader = null;
         List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
         for (CreateFuncParser.SqlParserResult funcInfo : funcList) {
             //classloader
             if (classLoader == null) {
-                classLoader = FlinkUtil.loadExtraJar(jarURList, (URLClassLoader)superClassLoader);
+                classLoader = FlinkUtil.loadExtraJar(jarURList, (URLClassLoader)levelClassLoader);
             }
             FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(),
                     tableEnv, classLoader);