[FLINK-38338][runtime] Introduce the abstraction to describe a rescale event.

Author: RocMarshal

flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/SlotSharingGroup.java

@@ -41,6 +41,8 @@ public class SlotSharingGroup implements java.io.Serializable {
 
     private final SlotSharingGroupId slotSharingGroupId = new SlotSharingGroupId();
 
+    private String slotSharingGroupName;
+
     // Represents resources of all tasks in the group. Default to be UNKNOWN.
     private ResourceProfile resourceProfile = ResourceProfile.UNKNOWN;
 
@@ -70,6 +72,14 @@ public ResourceProfile getResourceProfile() {
         return resourceProfile;
     }
 
+    public String getSlotSharingGroupName() {
+        return slotSharingGroupName;
+    }
+
+    public void setSlotSharingGroupName(String slotSharingGroupName) {
+        this.slotSharingGroupName = slotSharingGroupName;
+    }
+
     // ------------------------------------------------------------------------
     //  Utilities
     // ------------------------------------------------------------------------

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Finished.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.Durable;
 
 import org.slf4j.Logger;
 
@@ -34,13 +35,21 @@ class Finished implements State {
 
     private final Logger logger;
 
+    private final Durable durable;
+
     Finished(Context context, ArchivedExecutionGraph archivedExecutionGraph, Logger logger) {
         this.archivedExecutionGraph = archivedExecutionGraph;
         this.logger = logger;
+        this.durable = new Durable();
 
         context.onFinished(archivedExecutionGraph);
     }
 
+    @Override
+    public Durable getDurable() {
+        return durable;
+    }
+
     @Override
     public void cancel() {}
 

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.runtime.scheduler.adaptive;
 
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobgraph.JobVertex;
@@ -38,7 +39,7 @@ public class JobGraphJobInformation implements JobInformation {
     private final JobGraph jobGraph;
     private final JobID jobID;
     private final String name;
-    private final VertexParallelismStore vertexParallelismStore;
+    protected final VertexParallelismStore vertexParallelismStore;
 
     public JobGraphJobInformation(
             JobGraph jobGraph, VertexParallelismStore vertexParallelismStore) {
@@ -78,22 +79,29 @@ public Iterable<JobInformation.VertexInformation> getVertices() {
                 jobGraph.getVertices(), (vertex) -> getVertexInformation(vertex.getID()));
     }
 
+    @Override
+    public String getVertexName(JobVertexID jobVertexID) {
+        return jobGraph.findVertexByID(jobVertexID).getName();
+    }
+
     /** Returns a copy of a jobGraph that can be mutated. */
     public JobGraph copyJobGraph() {
         return InstantiationUtil.cloneUnchecked(jobGraph);
     }
 
+    @Override
     public VertexParallelismStore getVertexParallelismStore() {
         return vertexParallelismStore;
     }
 
-    private static final class JobVertexInformation implements JobInformation.VertexInformation {
+    @VisibleForTesting
+    public static final class JobVertexInformation implements JobInformation.VertexInformation {
 
         private final JobVertex jobVertex;
 
         private final VertexParallelismInformation parallelismInfo;
 
-        private JobVertexInformation(
+        public JobVertexInformation(
                 JobVertex jobVertex, VertexParallelismInformation parallelismInfo) {
             this.jobVertex = jobVertex;
             this.parallelismInfo = parallelismInfo;
@@ -123,5 +131,10 @@ public int getMaxParallelism() {
         public SlotSharingGroup getSlotSharingGroup() {
             return jobVertex.getSlotSharingGroup();
         }
+
+        @VisibleForTesting
+        public VertexParallelismInformation getVertexParallelismInfo() {
+            return parallelismInfo;
+        }
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/State.java

@@ -21,26 +21,37 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.Durable;
 import org.apache.flink.util.function.FunctionWithException;
 import org.apache.flink.util.function.ThrowingConsumer;
 
 import org.slf4j.Logger;
 
+import java.time.Instant;
 import java.util.Optional;
 import java.util.function.Consumer;
 
 /**
  * State abstraction of the {@link AdaptiveScheduler}. This interface contains all methods every
  * state implementation must support.
  */
-interface State extends LabeledGlobalFailureHandler {
+public interface State extends LabeledGlobalFailureHandler {
+
+    /**
+     * Get the durable time information of the current state.
+     *
+     * @return The durable time information of the current state.
+     */
+    Durable getDurable();
 
     /**
      * This method is called whenever one transitions out of this state.
      *
      * @param newState newState is the state into which the scheduler transitions
      */
-    default void onLeave(Class<? extends State> newState) {}
+    default void onLeave(Class<? extends State> newState) {
+        getDurable().setOutTimestamp(Instant.now().toEpochMilli());
+    }
 
     /** Cancels the job execution. */
     void cancel();

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithExecutionGraph.java

@@ -55,6 +55,7 @@
 import org.apache.flink.runtime.scheduler.KvStateHandler;
 import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
 import org.apache.flink.runtime.scheduler.VertexEndOfDataListener;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.Durable;
 import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry;
 import org.apache.flink.runtime.scheduler.exceptionhistory.RootExceptionHistoryEntry;
 import org.apache.flink.runtime.scheduler.stopwithsavepoint.StopWithSavepointTerminationManager;
@@ -69,6 +70,7 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -101,6 +103,8 @@ abstract class StateWithExecutionGraph implements State {
 
     private final VertexEndOfDataListener vertexEndOfDataListener;
 
+    private final Durable durable;
+
     StateWithExecutionGraph(
             Context context,
             ExecutionGraph executionGraph,
@@ -118,6 +122,7 @@ abstract class StateWithExecutionGraph implements State {
         this.userCodeClassLoader = userClassCodeLoader;
         this.failureCollection = new ArrayList<>(failureCollection);
         this.vertexEndOfDataListener = new VertexEndOfDataListener(executionGraph);
+        this.durable = new Durable();
 
         FutureUtils.assertNoException(
                 executionGraph
@@ -137,6 +142,11 @@ abstract class StateWithExecutionGraph implements State {
                                 context.getMainThreadExecutor()));
     }
 
+    @Override
+    public Durable getDurable() {
+        return durable;
+    }
+
     ExecutionGraph getExecutionGraph() {
         return executionGraph;
     }
@@ -156,6 +166,7 @@ protected ExecutionGraphHandler getExecutionGraphHandler() {
 
     @Override
     public void onLeave(Class<? extends State> newState) {
+        getDurable().setOutTimestamp(Instant.now().toEpochMilli());
         if (!StateWithExecutionGraph.class.isAssignableFrom(newState)) {
             // we are leaving the StateWithExecutionGraph --> we need to dispose temporary services
             operatorCoordinatorHandler.disposeAllOperatorCoordinators();

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateWithoutExecutionGraph.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.scheduler.adaptive.timeline.Durable;
 
 import org.slf4j.Logger;
 
@@ -39,9 +40,17 @@ abstract class StateWithoutExecutionGraph implements State {
 
     private final Logger logger;
 
+    private final Durable durable;
+
     StateWithoutExecutionGraph(Context context, Logger logger) {
         this.context = context;
         this.logger = logger;
+        this.durable = new Durable();
+    }
+
+    @Override
+    public Durable getDurable() {
+        return durable;
     }
 
     @Override

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/AllocatorUtil.java

@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.scheduler.adaptive.allocator;
 
-import org.apache.flink.runtime.instance.SlotSharingGroupId;
 import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
 import org.apache.flink.runtime.jobmaster.SlotInfo;
 import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
@@ -38,13 +37,13 @@ class AllocatorUtil {
 
     private AllocatorUtil() {}
 
-    static Map<SlotSharingGroupId, SlotSharingSlotAllocator.SlotSharingGroupMetaInfo>
+    static Map<SlotSharingGroup, SlotSharingSlotAllocator.SlotSharingGroupMetaInfo>
             getSlotSharingGroupMetaInfos(JobInformation jobInformation) {
         return SlotSharingSlotAllocator.SlotSharingGroupMetaInfo.from(jobInformation.getVertices());
     }
 
     static int getMinimumRequiredSlots(
-            Map<SlotSharingGroupId, SlotSharingSlotAllocator.SlotSharingGroupMetaInfo>
+            Map<SlotSharingGroup, SlotSharingSlotAllocator.SlotSharingGroupMetaInfo>
                     slotSharingGroupMetaInfos) {
         return slotSharingGroupMetaInfos.values().stream()
                 .map(SlotSharingSlotAllocator.SlotSharingGroupMetaInfo::getMaxLowerBound)
@@ -78,7 +77,10 @@ static void checkMinimumRequiredSlots(
                             }
                         });
         return sharedSlotToVertexAssignment.values().stream()
-                .map(SlotSharingSlotAllocator.ExecutionSlotSharingGroup::new)
+                .map(
+                        ids ->
+                                new SlotSharingSlotAllocator.ExecutionSlotSharingGroup(
+                                        slotSharingGroup, ids))
                 .collect(Collectors.toList());
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.scheduler.VertexParallelismStore;
 
 import java.util.Collection;
 
@@ -38,6 +39,14 @@ public interface JobInformation {
 
     Iterable<VertexInformation> getVertices();
 
+    default String getVertexName(JobVertexID jobVertexID) {
+        throw new UnsupportedOperationException();
+    }
+
+    default VertexParallelismStore getVertexParallelismStore() {
+        throw new UnsupportedOperationException();
+    }
+
     /** Information about a single vertex. */
     interface VertexInformation {
         JobVertexID getJobVertexID();