[FLINK-36011] [runtime] Generalize RescaleManager to become StateTran…

…sitionManager
dufeng1010 · Sep 2, 2024 · c869326 · c869326
1 parent 339f97c
commit c869326
Show file tree

Hide file tree

Showing 11 changed files with 1,250 additions and 975 deletions.
diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/...-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -349,7 +349,7 @@ public int getRescaleOnFailedCheckpointCount() {
     }
 
     private final Settings settings;
-    private final RescaleManager.Factory rescaleManagerFactory;
+    private final StateTransitionManager.Factory stateTransitionManagerFactory;
 
     private final JobGraph jobGraph;
 
@@ -427,7 +427,7 @@ public AdaptiveScheduler(
             throws JobExecutionException {
         this(
                 settings,
-                DefaultRescaleManager.Factory.fromSettings(settings),
+                DefaultStateTransitionManager.Factory.fromSettings(settings),
                 (metricGroup, checkpointStatsListener) ->
                         new DefaultCheckpointStatsTracker(
                                 configuration.get(WebOptions.CHECKPOINTS_HISTORY_SIZE),
@@ -455,7 +455,7 @@ public AdaptiveScheduler(
     @VisibleForTesting
     AdaptiveScheduler(
             Settings settings,
-            RescaleManager.Factory rescaleManagerFactory,
+            StateTransitionManager.Factory stateTransitionManagerFactory,
             BiFunction<JobManagerJobMetricGroup, CheckpointStatsListener, CheckpointStatsTracker>
                     checkpointStatsTrackerFactory,
             JobGraph jobGraph,
@@ -480,7 +480,7 @@ public AdaptiveScheduler(
         assertPreconditions(jobGraph);
 
         this.settings = settings;
-        this.rescaleManagerFactory = rescaleManagerFactory;
+        this.stateTransitionManagerFactory = stateTransitionManagerFactory;
 
         this.jobGraph = jobGraph;
         this.jobInfo = new JobInfoImpl(jobGraph.getJobID(), jobGraph.getName());
@@ -1175,7 +1175,7 @@ public void goToExecuting(
                         this,
                         userCodeClassLoader,
                         failureCollection,
-                        rescaleManagerFactory,
+                        stateTransitionManagerFactory,
                         settings.getMinParallelismChangeForDesiredRescale(),
                         settings.getRescaleOnFailedCheckpointCount()));
     }

diff --git a/...time/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManager.java b/...time/src/main/java/org/apache/flink/runtime/scheduler/adaptive/DefaultRescaleManager.java