Merge branch '1.8.0_dev_laucher' into 'v1.8.0_dev'

[LauncherMain启动依赖配置调整]



See merge request !72

Author: zoudaokoulife

core/src/main/java/com/dtstack/flink/sql/exec/FlinkSQLExec.java

@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.dtstack.flink.sql.exec;
 
 import org.apache.calcite.sql.SqlIdentifier;

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

@@ -30,7 +30,6 @@
 
 import java.io.File;
 import java.net.MalformedURLException;
-import java.net.URL;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
@@ -75,23 +74,22 @@ public AbstractYarnClusterDescriptor createPerJobClusterDescriptor(Properties co
 
         }
 
-        List<URL> classpaths = new ArrayList<>();
+         List<File> shipFiles = new ArrayList<>();
         if (flinkJarPath != null) {
             File[] jars = new File(flinkJarPath).listFiles();
 
             for (File file : jars){
                 if (file.toURI().toURL().toString().contains("flink-dist")){
                     clusterDescriptor.setLocalJarPath(new Path(file.toURI().toURL().toString()));
                 } else {
-                    classpaths.add(file.toURI().toURL());
+                    shipFiles.add(file);
                 }
             }
 
         } else {
             throw new RuntimeException("The Flink jar path is null");
         }
-
-        //clusterDescriptor.setProvidedUserJarFiles(classpaths);
+        clusterDescriptor.addShipFiles(shipFiles);
 
         if(!Strings.isNullOrEmpty(queue)){
             clusterDescriptor.setQueue(queue);

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

@@ -22,12 +22,11 @@
 import com.dtstack.flink.sql.util.PluginUtil;
 import org.apache.commons.io.Charsets;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.cache.DistributedCache;
 import org.apache.flink.client.deployment.ClusterSpecification;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.core.fs.Path;
 import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.shaded.guava18.com.google.common.base.Strings;
-import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
 import org.apache.flink.yarn.AbstractYarnClusterDescriptor;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.slf4j.Logger;
@@ -36,7 +35,10 @@
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLDecoder;
-import java.util.*;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
 
 /**
  * per job mode submitter
@@ -54,7 +56,7 @@ public static String submit(Options launcherOptions, JobGraph jobGraph) throws E
 		fillJobGraphClassPath(jobGraph);
 		if (!StringUtils.isBlank(launcherOptions.getAddjar())) {
 			String addjarPath = URLDecoder.decode(launcherOptions.getAddjar(), Charsets.UTF_8.toString());
-			List<String>  paths = getJarPaths(addjarPath);
+			List<String> paths = getJarPaths(addjarPath);
 			paths.forEach( path -> {
 				jobGraph.addJar(new Path("file://" + path));
 			});
@@ -94,31 +96,11 @@ private static List<String> getJarPaths(String addjarPath) {
 	}
 
 	private static void fillJobGraphClassPath(JobGraph jobGraph) throws MalformedURLException {
-		Map<String, String> jobCacheFileConfig = jobGraph.getJobConfiguration().toMap();
-		Set<String> classPathKeySet = Sets.newHashSet();
-
-		for(Map.Entry<String, String> tmp : jobCacheFileConfig.entrySet()){
-			if(Strings.isNullOrEmpty(tmp.getValue())){
-				continue;
-			}
-
-			if(tmp.getValue().startsWith("class_path")){
-				//DISTRIBUTED_CACHE_FILE_NAME_1
-				//DISTRIBUTED_CACHE_FILE_PATH_1
-				String key = tmp.getKey();
-				String[] array = key.split("_");
-				if(array.length < 5){
-					continue;
-				}
-
-				array[3] = "PATH";
-				classPathKeySet.add(StringUtils.join(array, "_"));
-			}
-		}
-
-		for(String key : classPathKeySet){
-			String pathStr = jobCacheFileConfig.get(key);
-			jobGraph.getClasspaths().add(new URL("file:" + pathStr));
-		}
+        Map<String, DistributedCache.DistributedCacheEntry> jobCacheFileConfig = jobGraph.getUserArtifacts();
+        for(Map.Entry<String,  DistributedCache.DistributedCacheEntry> tmp : jobCacheFileConfig.entrySet()){
+            if(tmp.getKey().startsWith("class_path")){
+                jobGraph.getClasspaths().add(new URL("file:" + tmp.getValue().filePath));
+            }
+        }
 	}
 }

rdb/rdb-side/src/main/java/com/dtstack/flink/sql/side/rdb/provider/DTC3P0DataSourceProvider.java

@@ -1,3 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package com.dtstack.flink.sql.side.rdb.provider;
 
 import com.mchange.v2.c3p0.ComboPooledDataSource;