Address the david's comments.

Author: RocMarshal

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

@@ -23,6 +23,7 @@
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 
 import java.util.Collections;
+import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
 
@@ -84,8 +85,23 @@ public void setSlotSharingGroupName(String slotSharingGroupName) {
     //  Utilities
     // ------------------------------------------------------------------------
 
+    @Override
+    public int hashCode() {
+        return Objects.hash(ids, slotSharingGroupId, slotSharingGroupName, resourceProfile);
+    }
+
     @Override
     public String toString() {
-        return "SlotSharingGroup{" + "ids=" + ids + ", resourceProfile=" + resourceProfile + '}';
+        return "SlotSharingGroup{"
+                + "ids="
+                + ids
+                + ", slotSharingGroupId="
+                + slotSharingGroupId
+                + ", slotSharingGroupName='"
+                + slotSharingGroupName
+                + '\''
+                + ", resourceProfile="
+                + resourceProfile
+                + '}';
     }
 }

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

@@ -79,11 +79,6 @@ 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);
@@ -112,6 +107,11 @@ public JobVertexID getJobVertexID() {
             return jobVertex.getID();
         }
 
+        @Override
+        public String getVertexName() {
+            return jobVertex.getName();
+        }
+
         @Override
         public int getMinParallelism() {
             return parallelismInfo.getMinParallelism();

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

@@ -39,10 +39,6 @@ public interface JobInformation {
 
     Iterable<VertexInformation> getVertices();
 
-    default String getVertexName(JobVertexID jobVertexID) {
-        throw new UnsupportedOperationException();
-    }
-
     default VertexParallelismStore getVertexParallelismStore() {
         throw new UnsupportedOperationException();
     }
@@ -51,6 +47,8 @@ default VertexParallelismStore getVertexParallelismStore() {
     interface VertexInformation {
         JobVertexID getJobVertexID();
 
+        String getVertexName();
+
         int getMinParallelism();
 
         int getParallelism();

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

@@ -22,7 +22,7 @@
 
 import java.time.Instant;
 
-/** Durable to record enter timestamp and leave timestamp. */
+/** Durable to record in timestamp and out timestamp. */
 public class Durable {
 
     private final Long inTimestamp;

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

@@ -51,7 +51,65 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
-/** Rescale event. */
+/**
+ * The rescale to record the related vertices and slots change during the rescaling process.
+ *
+ * <p>This rescale begins when the scheduler initiates a rescaling operation and ends when the
+ * rescaling succeeds.
+ *
+ * <pre>
+ *
+ * The structure of the rescale as follows:
+ *
+ * +--> rescale id information:
+ * +    +-->rescale uuid
+ * +    +-->resource requirements id
+ * +    +-->rescale attempt id
+ * +--> vertices:
+ * +    +--> job vertex id-1 -> vertex-1 parallelism rescale:
+ * +    +                       +--> vertex id
+ * +    +                       +--> vertex name
+ * +    +                       +--> slot sharing group id
+ * +    +                       +--> slot sharing group name
+ * +    +                       +--> desired parallelism
+ * +    +                       +--> sufficient parallelism
+ * +    +--> job vertex id-2 -> vertex-2 parallelism rescale:
+ * +    +                       +--> ...
+ * +    +                       ...
+ * +    ...
+ * +--> slots:
+ * +    +--> slot sharing group id-1 -> slot-1 sharing group rescale:
+ * +    +                               +--> slot sharing group id
+ * +    +                               +--> slot sharing group name
+ * +    +                               +--> required resource profile
+ * +    +                               +--> minimal required slots
+ * +    +                               +--> pre-rescale slots
+ * +    +                               +--> post-rescale slots
+ * +    +                               +--> acquired resource profile
+ * +    +--> slot sharing group id-2 -> slot-2 sharing group rescale:
+ * +    +                               +--> ...
+ * +    +                               ...
+ * +    ...
+ * +--> scheduler states:
+ * +    +--> scheduler state span:
+ * +    +    +--> state
+ * +    +    +--> in timestamp
+ * +    +    +--> out timestamp
+ * +    +    +--> duration
+ * +    +    +--> exception information
+ * +    +--> ...
+ * +    ...
+ * +--> start timestamp
+ * +--> end timestamp
+ * +--> trigger cause
+ * +--> terminal state
+ * +--> terminated reason
+ *
+ * </pre>
+ *
+ * <p>The more design details about the rescale could be viewed in <a
+ * href="https://cwiki.apache.org/confluence/x/TQr0Ew">FLIP-495</a>.
+ */
 public class Rescale implements Serializable {
 
     private static final long serialVersionUID = 1L;
@@ -167,6 +225,10 @@ public Rescale setStartTimestamp(long timestamp) {
         return this;
     }
 
+    public Long getStartTimestamp() {
+        return startTimestamp;
+    }
+
     public Rescale setEndTimestamp(Long endTimestamp) {
         if (this.endTimestamp != null) {
             LOG.warn("The old endTimestamp was already set to '{}'", this.endTimestamp);
@@ -175,6 +237,10 @@ public Rescale setEndTimestamp(Long endTimestamp) {
         return this;
     }
 
+    public Long getEndTimestamp() {
+        return endTimestamp;
+    }
+
     public Rescale setDesiredSlots(JobInformation jobInformation) {
         for (SlotSharingGroup sharingGroup : jobInformation.getSlotSharingGroups()) {
             int desiredSlot =
@@ -186,10 +252,10 @@ public Rescale setDesiredSlots(JobInformation jobInformation) {
                                                     .getParallelism())
                             .max(Integer::compare)
                             .orElse(0);
-            SlotSharingGroupId sharingGroupId = sharingGroup.getSlotSharingGroupId();
             SlotSharingGroupRescale sharingGroupRescaleInfo =
-                    slots.computeIfAbsent(sharingGroupId, SlotSharingGroupRescale::new);
-            sharingGroupRescaleInfo.setSlotSharingGroupMetaInfo(sharingGroup);
+                    slots.computeIfAbsent(
+                            sharingGroup.getSlotSharingGroupId(),
+                            ignored -> new SlotSharingGroupRescale(sharingGroup));
             sharingGroupRescaleInfo.setDesiredSlots(desiredSlot);
         }
         return this;
@@ -201,15 +267,16 @@ public Rescale setDesiredVertexParallelism(JobInformation jobInformation) {
         for (Map.Entry<JobVertexID, VertexParallelismInformation> entry :
                 allParallelismInfo.entrySet()) {
             JobVertexID jvId = entry.getKey();
-            VertexParallelismInformation vertexParallelInfo = entry.getValue();
-            VertexParallelismRescale vertexParallelismRescale =
-                    this.vertices.computeIfAbsent(
-                            jvId, jobVertexID -> new VertexParallelismRescale(jvId));
             SlotSharingGroup slotSharingGroup =
                     jobInformation.getVertexInformation(jvId).getSlotSharingGroup();
-            vertexParallelismRescale.setSlotSharingGroupMetaInfo(slotSharingGroup);
-            vertexParallelismRescale.setJobVertexName(jobInformation.getVertexName(jvId));
-            vertexParallelismRescale.setRequiredParallelisms(vertexParallelInfo);
+            String vertexName = jobInformation.getVertexInformation(jvId).getVertexName();
+            VertexParallelismRescale vertexParallelismRescale =
+                    this.vertices.computeIfAbsent(
+                            jvId,
+                            jobVertexID ->
+                                    new VertexParallelismRescale(
+                                            jvId, vertexName, slotSharingGroup));
+            vertexParallelismRescale.setRequiredParallelisms(entry.getValue());
         }
         return this;
     }
@@ -219,43 +286,69 @@ public Rescale setMinimalRequiredSlots(JobInformation jobInformation) {
                 SlotSharingGroupMetaInfo.from(jobInformation.getVertices());
         for (Map.Entry<SlotSharingGroup, SlotSharingGroupMetaInfo> entry :
                 slotSharingGroupMetaInfo.entrySet()) {
-            SlotSharingGroupId groupId = entry.getKey().getSlotSharingGroupId();
+            SlotSharingGroup sharingGroup = entry.getKey();
             SlotSharingGroupRescale slotSharingGroupRescale =
-                    slots.computeIfAbsent(groupId, SlotSharingGroupRescale::new);
+                    slots.computeIfAbsent(
+                            sharingGroup.getSlotSharingGroupId(),
+                            ignored -> new SlotSharingGroupRescale(sharingGroup));
             slotSharingGroupRescale.setMinimalRequiredSlots(entry.getValue().getMaxLowerBound());
         }
         return this;
     }
 
-    public Rescale setPreRescaleSlotsAndParallelisms(@Nullable Rescale lastCompletedRescale) {
+    public Rescale setPreRescaleSlotsAndParallelisms(
+            JobInformation jobInformation, @Nullable Rescale lastCompletedRescale) {
         if (lastCompletedRescale == null) {
             LOG.info("No available previous parallelism to set.");
             return this;
         }
         for (JobVertexID jobVertexID : lastCompletedRescale.getVertices().keySet()) {
             Integer preRescaleParallelism =
                     lastCompletedRescale.vertices.get(jobVertexID).getPostRescaleParallelism();
+            JobInformation.VertexInformation vertexInformation =
+                    jobInformation.getVertexInformation(jobVertexID);
             VertexParallelismRescale vertexParallelismRescale =
-                    vertices.computeIfAbsent(jobVertexID, VertexParallelismRescale::new);
+                    vertices.computeIfAbsent(
+                            jobVertexID,
+                            jobVertexId ->
+                                    new VertexParallelismRescale(
+                                            jobVertexId,
+                                            vertexInformation.getVertexName(),
+                                            vertexInformation.getSlotSharingGroup()));
             vertexParallelismRescale.setPreRescaleParallelism(preRescaleParallelism);
         }
 
-        for (SlotSharingGroupId sharingGroupId : lastCompletedRescale.getSlots().keySet()) {
-            Integer preRescaleSlot =
-                    lastCompletedRescale.slots.get(sharingGroupId).getPostRescaleSlots();
+        Map<SlotSharingGroupId, SlotSharingGroupRescale> slotsRescales =
+                lastCompletedRescale.getSlots();
+        for (SlotSharingGroup sharingGroup : jobInformation.getSlotSharingGroups()) {
+            SlotSharingGroupId slotSharingGroupId = sharingGroup.getSlotSharingGroupId();
+            Integer preRescaleSlot = slotsRescales.get(slotSharingGroupId).getPostRescaleSlots();
             SlotSharingGroupRescale slotSharingGroupRescale =
-                    slots.computeIfAbsent(sharingGroupId, SlotSharingGroupRescale::new);
+                    slots.computeIfAbsent(
+                            slotSharingGroupId,
+                            ignored -> new SlotSharingGroupRescale(sharingGroup));
             slotSharingGroupRescale.setPreRescaleSlots(preRescaleSlot);
         }
 
         return this;
     }
 
-    public Rescale setPostRescaleVertexParallelism(VertexParallelism postRescaleVertexParallelism) {
+    public Rescale setPostRescaleVertexParallelism(
+            JobInformation jobInformation, VertexParallelism postRescaleVertexParallelism) {
+
         Set<JobVertexID> vertices = postRescaleVertexParallelism.getVertices();
         for (JobVertexID vertexID : vertices) {
+            JobInformation.VertexInformation vertexInformation =
+                    jobInformation.getVertexInformation(vertexID);
             VertexParallelismRescale vertexParallelismRescale =
-                    this.vertices.computeIfAbsent(vertexID, VertexParallelismRescale::new);
+                    this.vertices.computeIfAbsent(
+                            vertexID,
+                            jobVertexId ->
+                                    new VertexParallelismRescale(
+                                            jobVertexId,
+                                            vertexInformation.getVertexName(),
+                                            vertexInformation.getSlotSharingGroup()));
+
             vertexParallelismRescale.setPostRescaleParallelism(
                     postRescaleVertexParallelism.getParallelism(vertexID));
         }
@@ -264,27 +357,28 @@ public Rescale setPostRescaleVertexParallelism(VertexParallelism postRescaleVert
 
     public Rescale setPostRescaleSlots(
             Collection<JobSchedulingPlan.SlotAssignment> postRescaleSlotAssignments) {
-        Map<SlotSharingGroupId, Set<JobSchedulingPlan.SlotAssignment>> assignmentsPerSharingGroup =
+        Map<SlotSharingGroup, Set<JobSchedulingPlan.SlotAssignment>> assignmentsPerSharingGroup =
                 postRescaleSlotAssignments.stream()
                         .collect(
                                 Collectors.groupingBy(
                                         slotAssignment ->
                                                 slotAssignment
                                                         .getTargetAs(
                                                                 ExecutionSlotSharingGroup.class)
-                                                        .getSlotSharingGroup()
-                                                        .getSlotSharingGroupId(),
+                                                        .getSlotSharingGroup(),
                                         Collectors.toSet()));
-        for (Map.Entry<SlotSharingGroupId, Set<JobSchedulingPlan.SlotAssignment>> entry :
+        for (Map.Entry<SlotSharingGroup, Set<JobSchedulingPlan.SlotAssignment>> entry :
                 assignmentsPerSharingGroup.entrySet()) {
-            SlotSharingGroupId sharingGroupId = entry.getKey();
+            SlotSharingGroup sharingGroup = entry.getKey();
             Set<JobSchedulingPlan.SlotAssignment> assignments =
-                    assignmentsPerSharingGroup.get(sharingGroupId);
+                    assignmentsPerSharingGroup.get(sharingGroup);
             int postRescaleSlot = assignments.size();
             ResourceProfile acquiredResource =
                     assignments.iterator().next().getSlotInfo().getResourceProfile();
             SlotSharingGroupRescale slotSharingGroupRescale =
-                    slots.computeIfAbsent(sharingGroupId, SlotSharingGroupRescale::new);
+                    slots.computeIfAbsent(
+                            sharingGroup.getSlotSharingGroupId(),
+                            ignored -> new SlotSharingGroupRescale(sharingGroup));
             slotSharingGroupRescale.setPostRescaleSlots(postRescaleSlot);
             slotSharingGroupRescale.setAcquiredResourceProfile(acquiredResource);
         }
@@ -296,16 +390,20 @@ public Rescale setTriggerCause(TriggerCause triggerCause) {
         return this;
     }
 
+    public TriggerCause getTriggerCause() {
+        return triggerCause;
+    }
+
     public void log() {
         LOG.info("Updated rescale is: {}", this);
     }
 
     public Map<JobVertexID, VertexParallelismRescale> getVertices() {
-        return vertices;
+        return Collections.unmodifiableMap(vertices);
     }
 
     public Map<SlotSharingGroupId, SlotSharingGroupRescale> getSlots() {
-        return slots;
+        return Collections.unmodifiableMap(slots);
     }
 
     public static boolean isTerminated(Rescale rescale) {

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

@@ -25,7 +25,11 @@
 import java.io.Serializable;
 import java.util.Objects;
 
-/** Utils class to record the information of a scheduler state. */
+/**
+ * Utils class to record the information of a scheduler state that contains the span of the time of
+ * the adaptive scheduler state, in timestamp, out timestamp, the exception if occurred during the
+ * adaptive scheduler state.
+ */
 public class SchedulerStateSpan implements Serializable {
     private static final long serialVersionUID = 1L;
 
@@ -61,6 +65,20 @@ public void setStringedException(@Nullable String stringedException) {
         this.stringedException = stringedException;
     }
 
+    public String getState() {
+        return state;
+    }
+
+    @Nullable
+    public Long getInTimestamp() {
+        return inTimestamp;
+    }
+
+    @Nullable
+    public Long getDuration() {
+        return duration;
+    }
+
     @Nullable
     public String getStringedException() {
         return stringedException;