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

Author: todd5167

README.md

@@ -50,10 +50,11 @@
 * 操作系统：理论上不限
 * kerberos环境需要在flink-conf.yaml配置security.kerberos.login.keytab以及security.kerberos.login.principal参数，配置案例:
 ```
+## hadoop配置文件路径
 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.keytab: /Users/maqi/tmp/hadoopconf/hadoop_250/maqi.keytab
+security.kerberos.login.principal: maqi@DTSTACK.COM
 security.kerberos.login.contexts: Client,KafkaClient
 zookeeper.sasl.service-name: zookeeper
 zookeeper.sasl.login-context-name: Client

core/src/main/java/com/dtstack/flink/sql/watermarker/AbsCustomerWaterMarker.java

@@ -21,13 +21,16 @@
 
 import com.dtstack.flink.sql.metric.EventDelayGauge;
 import com.dtstack.flink.sql.metric.MetricConstant;
+import com.dtstack.flink.sql.util.MathUtil;
 import org.apache.flink.api.common.functions.IterationRuntimeContext;
 import org.apache.flink.api.common.functions.RichFunction;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
 import org.apache.flink.streaming.api.windowing.time.Time;
 
+import java.util.TimeZone;
+
 /**
  * Reason:
  * Date: 2018/10/18
@@ -46,6 +49,12 @@ public abstract class AbsCustomerWaterMarker<T> extends BoundedOutOfOrdernessTim
 
     protected transient EventDelayGauge eventDelayGauge;
 
+    protected int pos;
+
+    protected long lastTime = 0;
+
+    protected TimeZone timezone;
+
     public AbsCustomerWaterMarker(Time maxOutOfOrderness) {
         super(maxOutOfOrderness);
     }
@@ -90,4 +99,13 @@ public void setRuntimeContext(RuntimeContext t) {
     public void setFromSourceTag(String fromSourceTag) {
         this.fromSourceTag = fromSourceTag;
     }
+
+    protected long getExtractTimestamp(Long extractTime){
+
+        lastTime = extractTime + timezone.getOffset(extractTime);
+
+        eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
+
+        return lastTime;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForLong.java

@@ -39,14 +39,6 @@ public class CustomerWaterMarkerForLong extends AbsCustomerWaterMarker<Row> {
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForLong.class);
 
-    private static final long serialVersionUID = 1L;
-
-    private int pos;
-
-    private long lastTime = 0;
-
-    private TimeZone timezone;
-
     public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos,String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
@@ -58,12 +50,7 @@ public long extractTimestamp(Row row) {
 
         try{
             Long extractTime = MathUtil.getLongVal(row.getField(pos));
-
-            lastTime = extractTime + timezone.getOffset(extractTime);
-
-            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
-
-            return lastTime;
+            return getExtractTimestamp(extractTime);
         }catch (Exception e){
             logger.error("", e);
         }

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForTimeStamp.java

@@ -40,14 +40,6 @@ public class CustomerWaterMarkerForTimeStamp extends AbsCustomerWaterMarker<Row>
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForTimeStamp.class);
 
-    private static final long serialVersionUID = 1L;
-
-    private int pos;
-
-    private long lastTime = 0;
-
-    private TimeZone timezone;
-
     public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos,String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
@@ -58,14 +50,8 @@ public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos,String ti
     public long extractTimestamp(Row row) {
         try {
             Timestamp time = (Timestamp) row.getField(pos);
-
             long extractTime=time.getTime();
-
-            lastTime = extractTime + timezone.getOffset(extractTime);
-
-            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
-
-            return lastTime;
+            return getExtractTimestamp(extractTime);
         } catch (RuntimeException e) {
             logger.error("", e);
         }

core/src/main/java/com/dtstack/flink/sql/watermarker/WaterMarkerAssigner.java

@@ -28,8 +28,8 @@
 import org.apache.flink.streaming.api.windowing.time.Time;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.Preconditions;
-
-import java.util.TimeZone;
+import java.sql.Timestamp;
+import java.lang.Long;
 
 /**
  * define watermarker
@@ -76,9 +76,9 @@ public DataStream assignWaterMarker(DataStream<Row> dataStream, RowTypeInfo type
         TypeInformation fieldType = fieldTypes[pos];
 
         AbsCustomerWaterMarker waterMarker = null;
-        if(fieldType.getTypeClass().getTypeName().equalsIgnoreCase("java.sql.Timestamp")){
+        if(fieldType.getTypeClass().isAssignableFrom(Timestamp.class)){
             waterMarker = new CustomerWaterMarkerForTimeStamp(Time.milliseconds(maxOutOrderness), pos,timeZone);
-        }else if(fieldType.getTypeClass().getTypeName().equalsIgnoreCase("java.lang.Long")){
+        }else if(fieldType.getTypeClass().isAssignableFrom(Long.class)){
             waterMarker = new CustomerWaterMarkerForLong(Time.milliseconds(maxOutOrderness), pos,timeZone);
         }else{
             throw new IllegalArgumentException("not support type of " + fieldType + ", current only support(timestamp, long).");

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobClusterClientBuilder.java

@@ -26,6 +26,8 @@
 import org.apache.flink.configuration.Configuration;
 import com.google.common.base.Strings;
 import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.security.SecurityConfiguration;
+import org.apache.flink.runtime.security.SecurityUtils;
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.hadoop.fs.Path;
@@ -55,45 +57,40 @@ public class PerJobClusterClientBuilder {
 
     private static final Logger LOG = LoggerFactory.getLogger(PerJobClusterClientBuilder.class);
 
-    private static String KEYTAB = "security.kerberos.login.keytab";
-
-    private static String PRINCIPAL = "security.kerberos.login.principal";
+    private static final String DEFAULT_CONF_DIR = "./";
 
     private YarnClient yarnClient;
 
     private YarnConfiguration yarnConf;
 
-    public void init(String yarnConfDir, Properties conf) throws IOException {
+    private Configuration flinkConfig;
+
+    public void init(String yarnConfDir, Configuration flinkConfig, Properties userConf) throws Exception {
 
         if(Strings.isNullOrEmpty(yarnConfDir)) {
             throw new RuntimeException("parameters of yarn is required");
         }
+        userConf.forEach((key, val) -> flinkConfig.setString(key.toString(), val.toString()));
+        this.flinkConfig = flinkConfig;
+        SecurityUtils.install(new SecurityConfiguration(flinkConfig));
 
         yarnConf = YarnConfLoader.getYarnConf(yarnConfDir);
-
-        if (isKerberos(conf)){
-            String keytab = (String) conf.get(KEYTAB);
-            String principal = (String) conf.get(PRINCIPAL);
-            login(yarnConf, keytab, principal);
-        }
-
         yarnClient = YarnClient.createYarnClient();
         yarnClient.init(yarnConf);
         yarnClient.start();
 
         System.out.println("----init yarn success ----");
     }
 
-    public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(Properties confProp, String flinkJarPath, Options launcherOptions, JobGraph jobGraph, Configuration flinkConfig)
+    public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(String flinkJarPath, Options launcherOptions, JobGraph jobGraph)
             throws MalformedURLException {
 
-        confProp.forEach((key, val) -> flinkConfig.setString(key.toString(), val.toString()));
-        String flinkConf = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? "" : launcherOptions.getFlinkconf();
+        String flinkConf = StringUtils.isEmpty(launcherOptions.getFlinkconf()) ? DEFAULT_CONF_DIR : launcherOptions.getFlinkconf();
         AbstractYarnClusterDescriptor clusterDescriptor = getClusterDescriptor(flinkConfig, yarnConf, flinkConf);
 
         if (StringUtils.isNotBlank(flinkJarPath)) {
             if (!new File(flinkJarPath).exists()) {
-                throw new RuntimeException("The Flink jar path is not exist");
+                throw new RuntimeException("The param '-flinkJarPath' ref dir is not exist");
             }
         }
 
@@ -163,22 +160,4 @@ private AbstractYarnClusterDescriptor getClusterDescriptor(
                 false);
     }
 
-    private boolean isKerberos(Properties conf){
-        String keytab = (String) conf.get(KEYTAB);
-        if (StringUtils.isNotBlank(keytab)){
-            return true;
-        } else {
-            return false;
-        }
-    }
-
-    private void login(org.apache.hadoop.conf.Configuration conf, String keytab, String principal) throws IOException {
-        if (StringUtils.isEmpty(principal)){
-            throw new RuntimeException(PRINCIPAL + " must not be null!");
-        }
-        UserGroupInformation.setConfiguration(conf);
-        UserGroupInformation.loginUserFromKeytab(principal, keytab);
-        LOG.info("login successfully! keytab: " + keytab + "principal: " + principal);
-        LOG.info("UGI: " + UserGroupInformation.getCurrentUser());
-    }
 }

launcher/src/main/java/com/dtstack/flink/sql/launcher/perjob/PerJobSubmitter.java

@@ -62,11 +62,10 @@ public static String submit(Options launcherOptions, JobGraph jobGraph, Configur
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
 
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
-        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), confProperties);
+        perJobClusterClientBuilder.init(launcherOptions.getYarnconf(), flinkConfig, confProperties);
 
         String flinkJarPath = launcherOptions.getFlinkJarPath();
-
-        AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(confProperties, flinkJarPath, launcherOptions, jobGraph, flinkConfig);
+        AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(flinkJarPath, launcherOptions, jobGraph);
         ClusterClient<ApplicationId> clusterClient = yarnClusterDescriptor.deployJobCluster(clusterSpecification, jobGraph,true);
 
         String applicationId = clusterClient.getClusterId().toString();