[Fix-127][core,executor] Add executeAsync datastream job

dlink-core/src/main/java/com/dlink/job/JobManager.java

@@ -35,6 +35,7 @@ import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.configuration.CoreOptions;
 import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
 import org.apache.flink.runtime.jobgraph.SavepointRestoreSettings;
@@ -355,9 +356,9 @@ public class JobManager {
                             break;
                         }
                     }
-                    JobExecutionResult jobExecutionResult = executor.execute(config.getJobName());
-                    if (jobExecutionResult.isJobExecutionResult()) {
-                        job.setJobId(jobExecutionResult.getJobID().toHexString());
+                    JobClient jobClient = executor.executeAsync(config.getJobName());
+                    if (Asserts.isNotNull(jobClient)) {
+                        job.setJobId(jobClient.getJobID().toHexString());
                         job.setJids(new ArrayList<String>() {{
                             add(job.getJobId());
                         }});

dlink-executor/src/main/java/com/dlink/executor/Executor.java

@@ -11,6 +11,7 @@ import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobExecutionResult;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.PipelineOptions;
+import org.apache.flink.core.execution.JobClient;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.jsonplan.JsonPlanGenerator;
 import org.apache.flink.runtime.rest.messages.JobPlanInfo;
@@ -206,6 +207,10 @@ public abstract class Executor {
         return environment.execute(jobName);
     }
 
+    public JobClient executeAsync(String jobName) throws Exception {
+        return environment.executeAsync(jobName);
+    }
+
     public TableResult executeSql(String statement) {
         statement = pretreatStatement(statement);
         FlinkInterceptorResult flinkInterceptorResult = pretreatExecute(statement);