session and submiit

Author: Mrart

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/deployment/YarnApplicationDeploymentExecutor.java

@@ -101,7 +101,7 @@ public PipelineExecution.ExecutionInfo deploy(
             if (client != null) {
                 client.shutDownCluster();
             }
-            throw new RuntimeException("Failed to deploy Flink CDC job", e);
+            throw new RuntimeException("Failed to yarn application deploy Flink CDC job", e);
         } finally {
             descriptor.close();
             if (client != null) {

flink-cdc-composer/src/main/java/org/apache/flink/cdc/composer/flink/deployment/YarnSessionDeploymentExecutor.java

@@ -22,11 +22,15 @@
 import org.apache.flink.client.deployment.application.ApplicationConfiguration;
 import org.apache.flink.client.program.ClusterClient;
 import org.apache.flink.client.program.ClusterClientProvider;
+import org.apache.flink.client.program.PackagedProgram;
+import org.apache.flink.client.program.PackagedProgramUtils;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.PipelineOptions;
 import org.apache.flink.core.fs.FSDataInputStream;
 import org.apache.flink.core.fs.FileSystem;
 import org.apache.flink.core.fs.Path;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
 import org.apache.flink.yarn.YarnClusterClientFactory;
 import org.apache.flink.yarn.YarnClusterDescriptor;
 import org.apache.flink.yarn.configuration.YarnConfigOptions;
@@ -38,10 +42,12 @@
 
 import org.apache.commons.cli.CommandLine;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -87,23 +93,72 @@ public PipelineExecution.ExecutionInfo deploy(
 
         ClusterClient<ApplicationId> client = null;
         try {
-            ClusterClientProvider<ApplicationId> clusterClientProvider =
-                    descriptor.deploySessionCluster(specification);
-            client = clusterClientProvider.getClusterClient();
-            ApplicationId clusterId = client.getClusterId();
-            LOG.info("Deployment Flink CDC From Cluster ID {}", clusterId);
-            return new PipelineExecution.ExecutionInfo(
-                    clusterId.toString(), "submit job successful");
+            // If applicationId is passed, we get the state of yarn; if not, we create a session
+            // cluster.
+            String applicationId = flinkConfig.get(YarnConfigOptions.APPLICATION_ID);
+            if (applicationId != null) {
+                FinalApplicationStatus applicationStatus =
+                        descriptor
+                                .getYarnClient()
+                                .getApplicationReport(ApplicationId.fromString(applicationId))
+                                .getFinalApplicationStatus();
+                if (FinalApplicationStatus.UNDEFINED.equals(applicationStatus)) {
+                    // applicationId is running.
+                    client =
+                            descriptor
+                                    .retrieve(ApplicationId.fromString(applicationId))
+                                    .getClusterClient();
+                }
+            } else {
+                ClusterClientProvider<ApplicationId> clusterClientProvider =
+                        descriptor.deploySessionCluster(specification);
+                client = clusterClientProvider.getClusterClient();
+                applicationId = String.valueOf(client.getClusterId());
+            }
+            LOG.info("Deployment Flink CDC From application ID {}", applicationId);
+            // how to get jobGraph
+            assert client != null;
+            client.submitJob(getJobGraph(flinkConfig, 1));
+
+            return new PipelineExecution.ExecutionInfo(applicationId, "submit job successful");
         } catch (Exception e) {
             if (client != null) {
                 client.shutDownCluster();
             }
-            throw new RuntimeException("Failed to deploy Flink CDC job", e);
+            throw new RuntimeException("Failed to yarn session deploy Flink CDC job", e);
         } finally {
             descriptor.close();
             if (client != null) {
                 client.close();
             }
         }
     }
+
+    /** Get jobGraph from configuration. */
+    private JobGraph getJobGraph(Configuration configuration, int parallelism) throws Exception {
+        SavepointRestoreSettings savepointRestoreSettings =
+                SavepointRestoreSettings.fromConfiguration(configuration);
+        PackagedProgram.Builder builder =
+                PackagedProgram.newBuilder()
+                        .setSavepointRestoreSettings(savepointRestoreSettings)
+                        .setEntryPointClassName(
+                                configuration
+                                        .getOptional(
+                                                ApplicationConfiguration.APPLICATION_MAIN_CLASS)
+                                        .get())
+                        .setArguments(
+                                configuration
+                                        .getOptional(ApplicationConfiguration.APPLICATION_ARGS)
+                                        .orElse(new ArrayList<>())
+                                        .toArray(new String[] {}))
+                        .setJarFile(
+                                new File(
+                                        configuration
+                                                .getOptional(PipelineOptions.JARS)
+                                                .orElse(new ArrayList<>())
+                                                .get(0)));
+        PackagedProgram program = builder.build();
+        return PackagedProgramUtils.createJobGraph(
+                program, configuration, parallelism, null, false);
+    }
 }