Merge remote-tracking branch 'origin/v1.5.0_dev' into v1.8.0_dev

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

Author: simenliuxing

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

@@ -76,6 +76,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;
@@ -213,11 +214,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

@@ -16,13 +16,12 @@
  * limitations under the License.
  */
 
-
+ 
 
 package com.dtstack.flink.sql.util;
 
 
 import com.dtstack.flink.sql.classloader.ClassLoaderManager;
-import com.dtstack.flink.sql.constrant.ConfigConstrant;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.runtime.state.filesystem.FsStateBackend;
@@ -35,8 +34,6 @@
 import org.apache.flink.table.api.java.StreamTableEnvironment;
 import org.apache.flink.table.functions.ScalarFunction;
 import org.apache.flink.table.functions.TableFunction;
-import org.apache.flink.table.functions.AggregateFunction;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -71,14 +68,12 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
         }
 
         //设置了时间间隔才表明开启了checkpoint
-        if(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY) == null && properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY) == null){
+        if(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY) == null){
             return;
         }else{
-            Long sql_interval = Long.valueOf(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY,"0"));
-            Long flink_interval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY, "0"));
-            long checkpointInterval = Math.max(sql_interval, flink_interval);
+            Long interval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY));
             //start checkpoint every ${interval}
-            env.enableCheckpointing(checkpointInterval);
+            env.enableCheckpointing(interval);
         }
 
         String checkMode = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_MODE_KEY);
@@ -106,14 +101,7 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
             env.getCheckpointConfig().setMaxConcurrentCheckpoints(maxConcurrCheckpoints);
         }
 
-        Boolean sqlCleanMode = MathUtil.getBoolean(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_CLEANUPMODE_KEY), false);
-        Boolean flinkCleanMode = MathUtil.getBoolean(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY), false);
-
-        String cleanupModeStr = "false";
-        if (sqlCleanMode || flinkCleanMode ){
-            cleanupModeStr = "true";
-        }
-
+        String cleanupModeStr = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY);
         if ("true".equalsIgnoreCase(cleanupModeStr)){
             env.getCheckpointConfig().enableExternalizedCheckpoints(
                     CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
@@ -149,7 +137,6 @@ public static void setStreamTimeCharacteristic(StreamExecutionEnvironment env, P
             if(characteristicStr.equalsIgnoreCase(tmp.toString())){
                 env.setStreamTimeCharacteristic(tmp);
                 flag = true;
-                break;
             }
         }
 
@@ -159,20 +146,21 @@ public static void setStreamTimeCharacteristic(StreamExecutionEnvironment env, P
     }
 
 
+
     /**
-     * TABLE|SCALA|AGGREGATE
+     * FIXME 暂时不支持 UDF 实现类--有参构造方法
+     * TABLE|SCALA
      * 注册UDF到table env
      */
     public static void registerUDF(String type, String classPath, String funcName, TableEnvironment tableEnv, List<URL> jarURList){
         if("SCALA".equalsIgnoreCase(type)){
             registerScalaUDF(classPath, funcName, tableEnv, jarURList);
         }else if("TABLE".equalsIgnoreCase(type)){
             registerTableUDF(classPath, funcName, tableEnv, jarURList);
-        }else if("AGGREGATE".equalsIgnoreCase(type)){
-            registerAggregateUDF(classPath, funcName, tableEnv, jarURList);
         }else{
-            throw new RuntimeException("not support of UDF which is not in (TABLE, SCALA, AGGREGATE)");
+            throw new RuntimeException("not support of UDF which is not in (TABLE, SCALA)");
         }
+
     }
 
     /**
@@ -194,7 +182,7 @@ public static void registerScalaUDF(String classPath, String funcName, TableEnvi
 
     /**
      * 注册自定义TABLEFFUNC方法到env上
-     *
+     * TODO 对User-Defined Aggregate Functions的支持
      * @param classPath
      * @param funcName
      * @param tableEnv
@@ -217,31 +205,6 @@ public static void registerTableUDF(String classPath, String funcName, TableEnvi
         }
     }
 
-    /**
-     * 注册自定义Aggregate FUNC方法到env上
-     *
-     * @param classPath
-     * @param funcName
-     * @param tableEnv
-     */
-    public static void registerAggregateUDF(String classPath, String funcName, TableEnvironment tableEnv, List<URL> jarURList) {
-        try {
-            AggregateFunction udfFunc = ClassLoaderManager.newInstance(jarURList, (cl) -> cl.loadClass(classPath).asSubclass(AggregateFunction.class).newInstance());
-
-            if (tableEnv instanceof StreamTableEnvironment) {
-                ((StreamTableEnvironment) tableEnv).registerFunction(funcName,  udfFunc);
-            } else if (tableEnv instanceof BatchTableEnvironment) {
-                ((BatchTableEnvironment) tableEnv).registerFunction(funcName,  udfFunc);
-            } else {
-                throw new RuntimeException("no support tableEnvironment class for " + tableEnv.getClass().getName());
-            }
-
-            logger.info("register Aggregate function:{} success.", funcName);
-        } catch (Exception e) {
-            logger.error("", e);
-            throw new RuntimeException("register Aggregate UDF exception:", e);
-        }
-    }
 
     /**
      *
@@ -276,9 +239,21 @@ public static long getBufferTimeoutMillis(Properties properties){
     }
 
     public static URLClassLoader loadExtraJar(List<URL> jarURLList, URLClassLoader classLoader) throws NoSuchMethodException, IllegalAccessException, InvocationTargetException {
+
+        int size = 0;
+        for(URL url : jarURLList){
+            if(url.toString().endsWith(".jar")){
+                size++;
+            }
+        }
+
+        URL[] urlArray = new URL[size];
+        int i=0;
         for(URL url : jarURLList){
             if(url.toString().endsWith(".jar")){
+                urlArray[i] = url;
                 urlClassLoaderAddUrl(classLoader, url);
+                i++;
             }
         }