fix conflict

Author: simenliuxing

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

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
- 
+
 
 package com.dtstack.flink.sql.util;
 
@@ -33,9 +33,10 @@
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.java.BatchTableEnvironment;
 import org.apache.flink.table.api.java.StreamTableEnvironment;
-import org.apache.flink.table.functions.AggregateFunction;
 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;
 
@@ -70,12 +71,14 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
         }
 
         //设置了时间间隔才表明开启了checkpoint
-        if(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY) == null){
+        if(properties.getProperty(ConfigConstrant.SQL_CHECKPOINT_INTERVAL_KEY) == null && properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY) == null){
             return;
         }else{
-            Long interval = Long.valueOf(properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_INTERVAL_KEY));
+            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);
             //start checkpoint every ${interval}
-            env.enableCheckpointing(interval);
+            env.enableCheckpointing(checkpointInterval);
         }
 
         String checkMode = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_MODE_KEY);
@@ -103,7 +106,14 @@ public static void openCheckpoint(StreamExecutionEnvironment env, Properties pro
             env.getCheckpointConfig().setMaxConcurrentCheckpoints(maxConcurrCheckpoints);
         }
 
-        String cleanupModeStr = properties.getProperty(ConfigConstrant.FLINK_CHECKPOINT_CLEANUPMODE_KEY);
+        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";
+        }
+
         if ("true".equalsIgnoreCase(cleanupModeStr)){
             env.getCheckpointConfig().enableExternalizedCheckpoints(
                     CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
@@ -139,6 +149,7 @@ public static void setStreamTimeCharacteristic(StreamExecutionEnvironment env, P
             if(characteristicStr.equalsIgnoreCase(tmp.toString())){
                 env.setStreamTimeCharacteristic(tmp);
                 flag = true;
+                break;
             }
         }
 
@@ -148,7 +159,6 @@ public static void setStreamTimeCharacteristic(StreamExecutionEnvironment env, P
     }
 
 
-
     /**
      * TABLE|SCALA|AGGREGATE
      * 注册UDF到table env
@@ -269,21 +279,9 @@ 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++;
             }
         }
 
@@ -311,4 +309,4 @@ public static TypeInformation[] transformTypes(Class[] fieldTypes){
         return types;
     }
 
-}
+}