修复 yarn perjob模式class加载不到问题

Author: zoudaokoulife

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

@@ -20,15 +20,24 @@
 
 import com.dtstack.flink.sql.launcher.LauncherOptions;
 import com.dtstack.flink.sql.util.PluginUtil;
+import org.apache.commons.io.Charsets;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.flink.client.deployment.ClusterSpecification;
 import org.apache.flink.client.program.ClusterClient;
 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;
 import org.slf4j.LoggerFactory;
 
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLDecoder;
+import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 
 /**
  * per job mode submitter
@@ -43,7 +52,11 @@ public class PerJobSubmitter {
 
     public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph) throws Exception {
 
-        Properties confProperties = PluginUtil.jsonStrToObject(launcherOptions.getConfProp(), Properties.class);
+        fillJobGraphClassPath(jobGraph);
+
+        String confProp = launcherOptions.getConfProp();
+        confProp = URLDecoder.decode(confProp, Charsets.UTF_8.toString());
+        Properties confProperties = PluginUtil.jsonStrToObject(confProp, Properties.class);
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
 
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
@@ -63,4 +76,33 @@ public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph)
 
         return applicationId;
     }
+
+    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));
+        }
+    }
 }