Merge branch '1.10_release_4.0.x' into 1.10_release_4.1.x

Author: gituser

core/src/main/java/com/dtstack/flink/sql/classloader/ClassLoaderManager.java

@@ -20,10 +20,12 @@
 
 import com.dtstack.flink.sql.util.PluginUtil;
 import com.dtstack.flink.sql.util.ReflectionUtils;
+import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.FileInputStream;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.net.URL;
@@ -71,9 +73,19 @@ private static DtClassLoader retrieveClassLoad(String pluginJarPath) {
         });
     }
 
-    private static DtClassLoader retrieveClassLoad(List<URL> jarUrls) {
+    public static DtClassLoader retrieveClassLoad(List<URL> jarUrls) {
         jarUrls.sort(Comparator.comparing(URL::toString));
-        String jarUrlkey = StringUtils.join(jarUrls, "_");
+
+        List<String> jarMd5s = new ArrayList<>(jarUrls.size());
+        for (URL jarUrl : jarUrls) {
+            try (FileInputStream inputStream = new FileInputStream(jarUrl.getPath())){
+                String jarMd5 = DigestUtils.md5Hex(inputStream);
+                jarMd5s.add(jarMd5);
+            } catch (Exception e) {
+                throw new RuntimeException("Exceptions appears when read file:" + e);
+            }
+        }
+        String jarUrlkey = StringUtils.join(jarMd5s, "_");
         return pluginClassLoader.computeIfAbsent(jarUrlkey, k -> {
             try {
                 URL[] urls = jarUrls.toArray(new URL[jarUrls.size()]);

core/src/main/java/com/dtstack/flink/sql/exec/ExecuteProcessHelper.java

@@ -277,20 +277,10 @@ private static void sqlTranslation(String localSqlPluginPath,
     public static void registerUserDefinedFunction(SqlTree sqlTree, List<URL> jarUrlList, TableEnvironment tableEnv, boolean getPlan)
             throws IllegalAccessException, InvocationTargetException {
         // udf和tableEnv须由同一个类加载器加载
-        ClassLoader levelClassLoader = tableEnv.getClass().getClassLoader();
         ClassLoader currentClassLoader = Thread.currentThread().getContextClassLoader();
-        URLClassLoader classLoader = null;
+        URLClassLoader classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) currentClassLoader);
         List<CreateFuncParser.SqlParserResult> funcList = sqlTree.getFunctionList();
         for (CreateFuncParser.SqlParserResult funcInfo : funcList) {
-            // 构建plan的情况下，udf和tableEnv不需要是同一个类加载器
-            if (getPlan) {
-                classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) currentClassLoader);
-            }
-
-            //classloader
-            if (classLoader == null) {
-                classLoader = ClassLoaderManager.loadExtraJar(jarUrlList, (URLClassLoader) levelClassLoader);
-            }
             FunctionManager.registerUDF(funcInfo.getType(), funcInfo.getClassName(), funcInfo.getName(), tableEnv, classLoader);
         }
     }