Merge branch 'v1.5.0_dev_udfclassloader' into 'v1.5.0_dev'

udf类加载器问题



See merge request !96

Author: simenliuxing

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

@@ -77,6 +77,7 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.net.URL;
+import java.net.URLClassLoader;
 import java.net.URLDecoder;
 import java.util.List;
 import java.util.Map;
@@ -237,11 +238,18 @@ private static void addEnvClassPath(StreamExecutionEnvironment env, Set<URL> cla
     }
 
     private static void registerUDF(SqlTree sqlTree, List<URL> jarURList, StreamTableEnvironment tableEnv)
-            throws ClassNotFoundException, NoSuchMethodException, IllegalAccessException, InvocationTargetException {
+            throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
         //register urf
+        // udf和tableEnv须由同一个类加载器加载
+        ClassLoader levelClassLoader = tableEnv.getClass().getClassLoader();
+        URLClassLoader classLoader = null;
         List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
         for (CreateFuncParser.SqlParserResult funcInfo : funcList) {
-            FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(), tableEnv, jarURList);
+            //classloader
+            if (classLoader == null) {
+                classLoader = FlinkUtil.loadExtraJar(jarURList, (URLClassLoader)levelClassLoader);
+            }
+            FlinkUtil.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(), tableEnv, classLoader);
         }
     }
 

core/src/main/java/com/dtstack/flink/sql/util/FlinkUtil.java

@@ -152,11 +152,11 @@ public static void setStreamTimeCharacteristic(StreamExecutionEnvironment env, P
      * TABLE|SCALA
      * 注册UDF到table env
      */
-    public static void registerUDF(String type, String classPath, String funcName, TableEnvironment tableEnv, List<URL> jarURList){
+    public static void registerUDF(String type, String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader){
         if("SCALA".equalsIgnoreCase(type)){
-            registerScalaUDF(classPath, funcName, tableEnv, jarURList);
+            registerScalaUDF(classPath, funcName, tableEnv, classLoader);
         }else if("TABLE".equalsIgnoreCase(type)){
-            registerTableUDF(classPath, funcName, tableEnv, jarURList);
+            registerTableUDF(classPath, funcName, tableEnv, classLoader);
         }else{
             throw new RuntimeException("not support of UDF which is not in (TABLE, SCALA)");
         }
@@ -169,9 +169,10 @@ public static void registerUDF(String type, String classPath, String funcName, T
      * @param funcName
      * @param tableEnv
      */
-    public static void registerScalaUDF(String classPath, String funcName, TableEnvironment tableEnv, List<URL> jarURList){
+    public static void registerScalaUDF(String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader){
         try{
-            ScalarFunction udfFunc = ClassLoaderManager.newInstance(jarURList, (cl) -> cl.loadClass(classPath).asSubclass(ScalarFunction.class).newInstance());
+            ScalarFunction udfFunc = Class.forName(classPath, false, classLoader)
+                    .asSubclass(ScalarFunction.class).newInstance();
             tableEnv.registerFunction(funcName, udfFunc);
             logger.info("register scala function:{} success.", funcName);
         }catch (Exception e){
@@ -187,9 +188,10 @@ public static void registerScalaUDF(String classPath, String funcName, TableEnvi
      * @param funcName
      * @param tableEnv
      */
-    public static void registerTableUDF(String classPath, String funcName, TableEnvironment tableEnv, List<URL> jarURList){
+    public static void registerTableUDF(String classPath, String funcName, TableEnvironment tableEnv, ClassLoader classLoader){
         try {
-            TableFunction udfFunc = ClassLoaderManager.newInstance(jarURList, (cl) -> cl.loadClass(classPath).asSubclass(TableFunction.class).newInstance());
+            ScalarFunction udfFunc = Class.forName(classPath, false, classLoader)
+                    .asSubclass(ScalarFunction.class).newInstance();
             if(tableEnv instanceof StreamTableEnvironment){
                 ((StreamTableEnvironment)tableEnv).registerFunction(funcName, udfFunc);
             }else if(tableEnv instanceof BatchTableEnvironment){