[hotfix] Remove DefaultExecutionGraph's jobType

Author: huyuanfeng

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java

@@ -223,9 +223,6 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
     /** Current status of the job execution. */
     private volatile JobStatus state = JobStatus.CREATED;
 
-    /** The job type of the job execution. */
-    private final JobType jobType;
-
     /** A future that completes once the job has reached a terminal state. */
     private final CompletableFuture<JobStatus> terminationFuture = new CompletableFuture<>();
 
@@ -311,7 +308,6 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
     // --------------------------------------------------------------------------------------------
 
     public DefaultExecutionGraph(
-            JobType jobType,
             JobInformation jobInformation,
             ScheduledExecutorService futureExecutor,
             Executor ioExecutor,
@@ -334,7 +330,6 @@ public DefaultExecutionGraph(
             TaskDeploymentDescriptorFactory taskDeploymentDescriptorFactory,
             List<JobStatusChangedListener> jobStatusChangedListeners) {
 
-        this.jobType = jobType;
         this.executionGraphId = new ExecutionGraphID();
 
         this.jobInformation = checkNotNull(jobInformation);
@@ -645,7 +640,7 @@ public JobStatus getState() {
 
     @Override
     public JobType getJobType() {
-        return jobType;
+        return jobInformation.getJobType();
     }
 
     @Override

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java

@@ -39,7 +39,6 @@
 import org.apache.flink.runtime.executiongraph.failover.partitionrelease.PartitionGroupReleaseStrategyFactoryLoader;
 import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
 import org.apache.flink.runtime.jobgraph.JobGraph;
-import org.apache.flink.runtime.jobgraph.JobType;
 import org.apache.flink.runtime.jobgraph.JobVertex;
 import org.apache.flink.runtime.jobgraph.jsonplan.JsonPlanGenerator;
 import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
@@ -105,12 +104,11 @@ public static DefaultExecutionGraph buildGraph(
 
         final String jobName = jobGraph.getName();
         final JobID jobId = jobGraph.getJobID();
-        final JobType jobType = jobGraph.getJobType();
 
         final JobInformation jobInformation =
                 new JobInformation(
                         jobId,
-                        jobType,
+                        jobGraph.getJobType(),
                         jobName,
                         jobGraph.getSerializedExecutionConfig(),
                         jobGraph.getJobConfiguration(),
@@ -149,7 +147,6 @@ public static DefaultExecutionGraph buildGraph(
         // create a new execution graph, if none exists so far
         final DefaultExecutionGraph executionGraph =
                 new DefaultExecutionGraph(
-                        jobGraph.getJobType(),
                         jobInformation,
                         futureExecutor,
                         ioExecutor,