Merge branch 'v1.8.0_dev' of ssh://git.dtstack.cn:10022/dtstack/dt-center-flinkStreamSQL into v1.8.0_dev_field_name_ignore_case

Author: dengdejiushini

README.md

@@ -48,6 +48,17 @@
 * Java: JDK8及以上
 * Flink集群: 1.4,1.5,1.8（单机模式不需要安装Flink集群）
 * 操作系统：理论上不限
+* kerberos环境需要在flink-conf.yaml配置security.kerberos.login.keytab以及security.kerberos.login.principal参数，配置案例:
+```
+fs.hdfs.hadoopconf: /Users/maqi/tmp/hadoopconf/hadoop_250
+security.kerberos.login.use-ticket-cache: true
+security.kerberos.login.keytab: /Users/maqi/tmp/hadoopconf/hadoop_250/yanxi.keytab
+security.kerberos.login.principal: yanxi@DTSTACK.COM
+security.kerberos.login.contexts: Client,KafkaClient
+zookeeper.sasl.service-name: zookeeper
+zookeeper.sasl.login-context-name: Client
+
+```
 
 ### 1.3 打包
 

cassandra/cassandra-side/cassandra-all-side/src/main/java/com/dtstack/flink/sql/side/cassandra/CassandraAllReqRow.java

@@ -268,7 +268,7 @@ private void loadData(Map<String, List<Map<String, Object>>> tmpCache) throws SQ
                         LOG.warn("get conn fail, wait for 5 sec and try again, connInfo:" + connInfo);
                         Thread.sleep(5 * 1000);
                     } catch (InterruptedException e1) {
-                        e1.printStackTrace();
+                        LOG.error("", e1);
                     }
                 }
 

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

@@ -19,11 +19,15 @@
 package com.dtstack.flink.sql.classloader;
 
 import com.dtstack.flink.sql.util.PluginUtil;
+import com.dtstack.flink.sql.util.ReflectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
 import java.net.URL;
+import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Comparator;
@@ -91,4 +95,26 @@ public static List<URL> getClassPath() {
         }
         return classPaths;
     }
+
+
+
+    public static URLClassLoader loadExtraJar(List<URL> jarURLList, URLClassLoader classLoader)
+            throws  IllegalAccessException, InvocationTargetException {
+        for(URL url : jarURLList){
+            if(url.toString().endsWith(".jar")){
+                urlClassLoaderAddUrl(classLoader, url);
+            }
+        }
+        return classLoader;
+    }
+
+    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/classloader/ClassLoaderManager.java

@@ -53,9 +53,10 @@ public class ConfigConstrant {
     public static final String SQL_BUFFER_TIMEOUT_MILLIS = "sql.buffer.timeout.millis";
 
     public static final String FLINK_TIME_CHARACTERISTIC_KEY = "time.characteristic";
+    // default 200ms
+    public static final String AUTO_WATERMARK_INTERVAL_KEY = "autoWatermarkInterval";
 
     public static final String SQL_TTL_MINTIME = "sql.ttl.min";
-
     public static final String SQL_TTL_MAXTIME = "sql.ttl.max";
 
     public static final String STATE_BACKEND_KEY = "state.backend";

core/src/main/java/com/dtstack/flink/sql/constrant/ConfigConstrant.java

@@ -24,7 +24,14 @@
  */
 public enum ClusterMode {
 
-    local(0),standalone(1),yarn(2),yarnPer(3);
+    //run in local
+    local(0),
+    //submit job to standalone cluster
+    standalone(1),
+    //submit job to flink-session which is already run on yarn
+    yarn(2),
+    //submit job to yarn cluster as an application
+    yarnPer(3);
 
     private int type;