[FLINK-25973][runtime] Renamed ArchivedExecutionGraph.createFromInitializingJob into createSparseArchivedExecutionGraph

XComp · XComp · commit 5575c9ad2962 · 2022-02-16T09:45:03.000+01:00
The renaming became necessary because the method is now also used
in the CheckpointResourcesCleanupRunner where the job is not in
initialization phase anymore.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -452,7 +452,7 @@ public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout)
     public CompletableFuture<Acknowledge> submitFailedJob(
             JobID jobId, String jobName, Throwable exception) {
         final ArchivedExecutionGraph archivedExecutionGraph =
-                ArchivedExecutionGraph.createFromInitializingJob(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                         jobId,
                         jobName,
                         JobStatus.FAILED,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/CheckpointResourcesCleanupRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/cleanup/CheckpointResourcesCleanupRunner.java
@@ -214,7 +214,7 @@ private static JobStatus getJobStatus(JobResult jobResult) {
     private static ExecutionGraphInfo generateExecutionGraphInfo(
             JobResult jobResult, long initializationTimestamp) {
         return new ExecutionGraphInfo(
-                ArchivedExecutionGraph.createFromInitializingJob(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                         jobResult.getJobId(),
                         "unknown",
                         getJobStatus(jobResult),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java
@@ -331,10 +331,10 @@ public static ArchivedExecutionGraph createFrom(
     }
 
     /**
-     * Create a sparse ArchivedExecutionGraph for a job while it is still initializing. Most fields
-     * will be empty, only job status and error-related fields are set.
+     * Create a sparse ArchivedExecutionGraph for a job. Most fields will be empty, only job status
+     * and error-related fields are set.
      */
-    public static ArchivedExecutionGraph createFromInitializingJob(
+    public static ArchivedExecutionGraph createSparseArchivedExecutionGraph(
             JobID jobId,
             String jobName,
             JobStatus jobStatus,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobMasterServiceProcessFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/factories/DefaultJobMasterServiceProcessFactory.java
@@ -68,7 +68,7 @@ public JobID getJobId() {
     @Override
     public ArchivedExecutionGraph createArchivedExecutionGraph(
             JobStatus jobStatus, @Nullable Throwable cause) {
-        return ArchivedExecutionGraph.createFromInitializingJob(
+        return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                 jobId, jobName, jobStatus, cause, checkpointingSettings, initializationTimestamp);
     }
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -761,7 +761,7 @@ private VertexParallelism determineParallelism(SlotAllocator slotAllocator)
     @Override
     public ArchivedExecutionGraph getArchivedExecutionGraph(
             JobStatus jobStatus, @Nullable Throwable cause) {
-        return ArchivedExecutionGraph.createFromInitializingJob(
+        return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                 jobInformation.getJobID(),
                 jobInformation.getName(),
                 jobStatus,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherResourceCleanupTest.java
@@ -694,7 +694,7 @@ public TestingJobManagerRunner createJobManagerRunner(
                                             CompletableFuture.completedFuture(
                                                     new ExecutionGraphInfo(
                                                             ArchivedExecutionGraph
-                                                                    .createFromInitializingJob(
+                                                                    .createSparseArchivedExecutionGraph(
                                                                             jobGraph.getJobID(),
                                                                             jobGraph.getName(),
                                                                             JobStatus.RUNNING,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -526,7 +526,7 @@ public void testJobManagerRunnerInitializationFailureFailsJob() throws Exception
         testingJobManagerRunner.completeResultFuture(
                 JobManagerRunnerResult.forInitializationFailure(
                         new ExecutionGraphInfo(
-                                ArchivedExecutionGraph.createFromInitializingJob(
+                                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                                         jobId,
                                         jobGraph.getName(),
                                         JobStatus.FAILED,
@@ -687,7 +687,7 @@ public void testFatalErrorIfRecoveredJobsCannotBeStarted() throws Exception {
         testingJobManagerRunner.completeResultFuture(
                 JobManagerRunnerResult.forInitializationFailure(
                         new ExecutionGraphInfo(
-                                ArchivedExecutionGraph.createFromInitializingJob(
+                                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                                         jobId,
                                         jobGraph.getName(),
                                         JobStatus.FAILED,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
@@ -152,7 +152,7 @@ public void testSerialization() throws IOException, ClassNotFoundException {
     @Test
     public void testCreateFromInitializingJobForSuspendedJob() {
         final ArchivedExecutionGraph suspendedExecutionGraph =
-                ArchivedExecutionGraph.createFromInitializingJob(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                         new JobID(),
                         "TestJob",
                         JobStatus.SUSPENDED,
@@ -170,7 +170,7 @@ public void testCheckpointSettingsArchiving() {
                 CheckpointCoordinatorConfiguration.builder().build();
 
         final ArchivedExecutionGraph archivedGraph =
-                ArchivedExecutionGraph.createFromInitializingJob(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                         new JobID(),
                         "TestJob",
                         JobStatus.INITIALIZING,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java
@@ -52,7 +52,7 @@ public class DefaultJobMasterServiceProcessTest extends TestLogger {
     private static final Function<Throwable, ArchivedExecutionGraph>
             failedArchivedExecutionGraphFactory =
                     (throwable ->
-                            ArchivedExecutionGraph.createFromInitializingJob(
+                            ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                                     jobId, "test", JobStatus.FAILED, throwable, null, 1337));
 
     @Test
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterServiceLeadershipRunnerTest.java
@@ -259,7 +259,7 @@ public void testJobMasterCreationFailureCompletesJobManagerRunnerWithInitializat
     @Nonnull
     private ExecutionGraphInfo createFailedExecutionGraphInfo(FlinkException testException) {
         return new ExecutionGraphInfo(
-                ArchivedExecutionGraph.createFromInitializingJob(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                         jobGraph.getJobID(),
                         jobGraph.getName(),
                         JobStatus.FAILED,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java
@@ -69,7 +69,7 @@ private TestingJobManagerRunner(
 
         final ExecutionGraphInfo suspendedExecutionGraphInfo =
                 new ExecutionGraphInfo(
-                        ArchivedExecutionGraph.createFromInitializingJob(
+                        ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                                 jobId, "TestJob", JobStatus.SUSPENDED, null, null, 0L),
                         null);
         terminationFuture.whenComplete(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/factories/TestingJobMasterServiceProcessFactory.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/factories/TestingJobMasterServiceProcessFactory.java
@@ -64,7 +64,7 @@ public JobID getJobId() {
     @Override
     public ArchivedExecutionGraph createArchivedExecutionGraph(
             JobStatus jobStatus, @Nullable Throwable cause) {
-        return ArchivedExecutionGraph.createFromInitializingJob(
+        return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                 jobId, jobName, jobStatus, cause, null, initializationTimestamp);
     }
 
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/factories/TestingJobMasterServiceProcessFactoryOld.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/factories/TestingJobMasterServiceProcessFactoryOld.java
@@ -72,7 +72,7 @@ public JobID getJobId() {
     @Override
     public ArchivedExecutionGraph createArchivedExecutionGraph(
             JobStatus jobStatus, @Nullable Throwable cause) {
-        return ArchivedExecutionGraph.createFromInitializingJob(
+        return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                 jobId, "test-job", jobStatus, cause, null, System.currentTimeMillis());
     }
 
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/ExecutionGraphInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/ExecutionGraphInfoTest.java
@@ -36,7 +36,7 @@ public class ExecutionGraphInfoTest {
     @Test
     public void testExecutionGraphHistoryBeingDerivedFromFailedExecutionGraph() {
         final ArchivedExecutionGraph executionGraph =
-                ArchivedExecutionGraph.createFromInitializingJob(
+                ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                         new JobID(),
                         "test job name",
                         JobStatus.FAILED,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java
@@ -117,7 +117,7 @@ public void goToFinished(ArchivedExecutionGraph archivedExecutionGraph) {
         @Override
         public ArchivedExecutionGraph getArchivedExecutionGraph(
                 JobStatus jobStatus, @Nullable Throwable cause) {
-            return ArchivedExecutionGraph.createFromInitializingJob(
+            return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                     new JobID(), "testJob", jobStatus, cause, null, 0L);
         }
 
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
@@ -210,7 +210,7 @@ public void goToExecuting(
         @Override
         public ArchivedExecutionGraph getArchivedExecutionGraph(
                 JobStatus jobStatus, @Nullable Throwable cause) {
-            return ArchivedExecutionGraph.createFromInitializingJob(
+            return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(
                     new JobID(), "testJob", jobStatus, cause, null, 0L);
         }
 

Original file line number	Diff line number	Diff line change
`@@ -68,7 +68,7 @@ public JobID getJobId() {`
`68`	`68`	`@Override`
`69`	`69`	`public ArchivedExecutionGraph createArchivedExecutionGraph(`
`70`	`70`	`JobStatus jobStatus, @Nullable Throwable cause) {`
`71`		`- return ArchivedExecutionGraph.createFromInitializingJob(`
	`71`	`+ return ArchivedExecutionGraph.createSparseArchivedExecutionGraph(`
`72`	`72`	`jobId, jobName, jobStatus, cause, checkpointingSettings, initializationTimestamp);`
`73`	`73`	`}`
`74`	`74`	`}`