modify base flinx version to 1.5.3

Author: zoudaokoulife

elasticsearch5/elasticsearch5-sink/pom.xml

@@ -16,7 +16,7 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-elasticsearch5_2.11</artifactId>
-            <version>1.4.0</version>
+            <version>${flink.version}</version>
         </dependency>
 
         <dependency>

kafka09/kafka09-source/pom.xml

@@ -14,11 +14,6 @@
     <name>kafka09-source</name>
     <url>http://maven.apache.org</url>
 
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.4.0</flink.version>
-    </properties>
-
     <dependencies>
 
         <dependency>

kafka10/kafka10-source/pom.xml

@@ -13,11 +13,6 @@
     <name>kafka10-source</name>
     <url>http://maven.apache.org</url>
 
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.4.0</flink.version>
-    </properties>
-
     <dependencies>
 
         <dependency>

kafka11/kafka11-source/pom.xml

@@ -13,11 +13,6 @@
     <name>kafka11-source</name>
     <url>http://maven.apache.org</url>
 
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.4.0</flink.version>
-    </properties>
-
     <dependencies>
 
         <dependency>

launcher/src/main/java/com/dtstack/flink/sql/launcher/ClusterClientFactory.java

@@ -19,16 +19,19 @@
 package com.dtstack.flink.sql.launcher;
 
 import org.apache.commons.lang.StringUtils;
+import org.apache.flink.client.deployment.ClusterRetrieveException;
 import org.apache.flink.client.deployment.StandaloneClusterDescriptor;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.StandaloneClusterClient;
+import org.apache.flink.client.program.rest.RestClusterClient;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.flink.yarn.YarnClusterClient;
 import org.apache.flink.yarn.YarnClusterDescriptor;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.client.api.YarnClient;
@@ -55,7 +58,7 @@
  */
 public class ClusterClientFactory {
 
-    public static ClusterClient createClusterClient(Properties props) {
+    public static ClusterClient createClusterClient(Properties props) throws ClusterRetrieveException {
         String clientType = props.getProperty(OPTION_MODE);
         if(clientType.equals(ClusterMode.MODE_STANDALONE)) {
             return createStandaloneClient(props);
@@ -65,20 +68,20 @@ public static ClusterClient createClusterClient(Properties props) {
         throw new IllegalArgumentException("Unsupported cluster client type: ");
     }
 
-    public static StandaloneClusterClient createStandaloneClient(Properties props) {
+    public static RestClusterClient createStandaloneClient(Properties props) throws ClusterRetrieveException {
         String flinkConfDir = props.getProperty(LauncherOptions.OPTION_FLINK_CONF_DIR);
         Configuration config = GlobalConfiguration.loadConfiguration(flinkConfDir);
         StandaloneClusterDescriptor descriptor = new StandaloneClusterDescriptor(config);
-        StandaloneClusterClient clusterClient = descriptor.retrieve(null);
+        RestClusterClient clusterClient = descriptor.retrieve(null);
         clusterClient.setDetached(true);
         return clusterClient;
     }
 
-    public static YarnClusterClient createYarnClient(Properties props) {
+    public static ClusterClient createYarnClient(Properties props) {
         String flinkConfDir = props.getProperty(LauncherOptions.OPTION_FLINK_CONF_DIR);
         Configuration config = GlobalConfiguration.loadConfiguration(flinkConfDir);
         String yarnConfDir = props.getProperty(LauncherOptions.OPTION_YARN_CONF_DIR);
-        org.apache.hadoop.conf.Configuration yarnConf = new YarnConfiguration();
+        YarnConfiguration yarnConf = new YarnConfiguration();
         if(StringUtils.isNotBlank(yarnConfDir)) {
             try {
 
@@ -96,6 +99,7 @@ public boolean accept(File dir, String name) {
                             return false;
                         }
                     });
+
                     if(xmlFileList != null) {
                         for(File xmlFile : xmlFileList) {
                             yarnConf.addResource(xmlFile.toURI().toURL());
@@ -105,7 +109,7 @@ public boolean accept(File dir, String name) {
                     YarnClient yarnClient = YarnClient.createYarnClient();
                     yarnClient.init(yarnConf);
                     yarnClient.start();
-                    String applicationId = null;
+                    ApplicationId applicationId = null;
 
                     Set<String> set = new HashSet<>();
                     set.add("Apache Flink");
@@ -129,24 +133,22 @@ public boolean accept(File dir, String name) {
                         if(thisMemory > maxMemory || thisMemory == maxMemory && thisCores > maxCores) {
                             maxMemory = thisMemory;
                             maxCores = thisCores;
-                            applicationId = report.getApplicationId().toString();
+                            applicationId = report.getApplicationId();
                         }
 
                     }
 
-                    if(org.apache.commons.lang3.StringUtils.isEmpty(applicationId)) {
+                    if(org.apache.commons.lang3.StringUtils.isEmpty(applicationId.toString())) {
                         throw new RuntimeException("No flink session found on yarn cluster.");
                     }
 
-                    yarnClient.stop();
-
-                    AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(config, ".");
+                    AbstractYarnClusterDescriptor clusterDescriptor = new YarnClusterDescriptor(config, yarnConf, ".", yarnClient, false);
                     Field confField = AbstractYarnClusterDescriptor.class.getDeclaredField("conf");
                     confField.setAccessible(true);
                     haYarnConf(yarnConf);
                     confField.set(clusterDescriptor, yarnConf);
 
-                    YarnClusterClient clusterClient = clusterDescriptor.retrieve(applicationId);
+                    ClusterClient clusterClient = clusterDescriptor.retrieve(applicationId);
                     clusterClient.setDetached(true);
                     return clusterClient;
                 }

mysql/mysql-sink/pom.xml

@@ -14,11 +14,6 @@
     <name>mysql-sink</name>
     <url>http://maven.apache.org</url>
 
-    <properties>
-        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.4.0</flink.version>
-    </properties>
-
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>

pom.xml

@@ -22,7 +22,7 @@
 
   <properties>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-    <flink.version>1.4.0</flink.version>
+    <flink.version>1.5.3</flink.version>
   </properties>
 
   <build>