yarnper submit bugfix

Author: [email protected]

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

@@ -63,7 +63,7 @@ public static ClusterClient createClusterClient(LauncherOptions launcherOptions)
         String mode = launcherOptions.getMode();
         if(mode.equals(ClusterMode.standalone.name())) {
             return createStandaloneClient(launcherOptions);
-        } else if(mode.equals(ClusterMode.yarn.name()) || mode.equals(ClusterMode.yarnPer.name())) {
+        } else if(mode.equals(ClusterMode.yarn.name())) {
             return createYarnClient(launcherOptions,mode);
         }
         throw new IllegalArgumentException("Unsupported cluster client type: ");

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

@@ -20,14 +20,17 @@
 
 import com.dtstack.flink.sql.options.LauncherOptions;
 import com.dtstack.flink.sql.util.PluginUtil;
+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.runtime.jobgraph.JobGraph;
 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.util.Map;
 import java.util.Properties;
 
 /**
@@ -39,28 +42,33 @@
 
 public class PerJobSubmitter {
 
+    private final static String CLASS_FILE_NAME_PRESTR = "class_path";
+
     private static final Logger LOG = LoggerFactory.getLogger(PerJobSubmitter.class);
 
     public static String submit(LauncherOptions launcherOptions, JobGraph jobGraph) throws Exception {
-
         Properties confProperties = PluginUtil.jsonStrToObject(launcherOptions.getConfProp(), Properties.class);
         ClusterSpecification clusterSpecification = FLinkPerJobResourceUtil.createClusterSpecification(confProperties);
-
         PerJobClusterClientBuilder perJobClusterClientBuilder = new PerJobClusterClientBuilder();
         perJobClusterClientBuilder.init(launcherOptions.getYarnconf());
-
         String flinkJarPath = launcherOptions.getFlinkJarPath();
-
         AbstractYarnClusterDescriptor yarnClusterDescriptor = perJobClusterClientBuilder.createPerJobClusterDescriptor(confProperties, flinkJarPath, launcherOptions.getQueue());
-        ClusterClient<ApplicationId> clusterClient = yarnClusterDescriptor.deployJobCluster(clusterSpecification, jobGraph,true);
-
+        ClusterClient<ApplicationId> clusterClient = yarnClusterDescriptor.deployJobCluster(clusterSpecification,fillJobGraphClassPath(jobGraph),true);
         String applicationId = clusterClient.getClusterId().toString();
         String flinkJobId = jobGraph.getJobID().toString();
-
         String tips = String.format("deploy per_job with appId: %s, jobId: %s", applicationId, flinkJobId);
         System.out.println(tips);
         LOG.info(tips);
-
         return applicationId;
     }
+
+    private static JobGraph fillJobGraphClassPath(JobGraph jobGraph) throws MalformedURLException {
+        Map<String, DistributedCache.DistributedCacheEntry> jobCacheFileConfig = jobGraph.getUserArtifacts();
+        for(Map.Entry<String,  DistributedCache.DistributedCacheEntry> tmp : jobCacheFileConfig.entrySet()){
+            if(tmp.getKey().startsWith(CLASS_FILE_NAME_PRESTR)){
+                jobGraph.getClasspaths().add(new URL("file:" + tmp.getValue().filePath));
+            }
+        }
+        return jobGraph;
+    }
 }