Merge branch '1.5_v3.6.0_yarnperjob_udf_classload_bug' into 'v1.5.0_dev'

修复 yarn perjob模式下加载 udf jar 问题



See merge request !13

Author: yanxi

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

@@ -267,8 +267,13 @@ public static URLClassLoader loadExtraJar(List<URL> jarURLList, URLClassLoader c
         return classLoader;
     }
 
-    private static void urlClassLoaderAddUrl(URLClassLoader classLoader, URL url) throws NoSuchMethodException, InvocationTargetException, IllegalAccessException {
-        Method method = classLoader.getClass().getDeclaredMethod("addURL", URL.class);
+    private static void urlClassLoaderAddUrl(URLClassLoader classLoader, URL url) throws InvocationTargetException, IllegalAccessException {
+        Method method = ReflectionUtils.getDeclaredMethod(classLoader, "addURL", URL.class);
+
+        if(method == null){
+            throw new RuntimeException("can't not find declared method addURL, curr classLoader is " + classLoader.getClass());
+        }
+
         method.setAccessible(true);
         method.invoke(classLoader, url);
     }

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

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.util;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+
+/**
+ *
+ * Date: 2019/4/22
+ * Company: www.dtstack.com
+ * @author xuchao
+ */
+
+public class ReflectionUtils {
+
+    /**
+     * get declaredMethod util find
+     * @param object
+     * @param methodName
+     * @param parameterTypes
+     * @return
+     */
+    public static Method getDeclaredMethod(Object object, String methodName, Class<?> ... parameterTypes){
+        Method method = null ;
+
+        for(Class<?> clazz = object.getClass() ; clazz != Object.class ; clazz = clazz.getSuperclass()) {
+            try {
+                method = clazz.getDeclaredMethod(methodName, parameterTypes) ;
+                return method ;
+            } catch (Exception e) {
+                //do nothing then can get method from super Class
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * 循环向上转型, 获取对象的 DeclaredField
+     * @param object : 子类对象
+     * @param fieldName : 父类中的属性名
+     * @return 父类中的属性对象
+     */
+
+    public static Field getDeclaredField(Object object, String fieldName){
+        Field field = null ;
+
+        Class<?> clazz = object.getClass() ;
+
+        for(; clazz != Object.class ; clazz = clazz.getSuperclass()) {
+            try {
+                field = clazz.getDeclaredField(fieldName) ;
+                return field ;
+            } catch (Exception e) {
+                //do nothing then can get field from super Class
+            }
+        }
+
+        return null;
+    }
+
+}