[FLINK-38718][runtime] Display the number of tasks at slots level on the detailed TaskManager page.

Author: RocMarshal

docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

@@ -5441,6 +5441,9 @@
           "jobId" : {
             "type" : "any"
           },
+          "numberOfTasks" : {
+            "type" : "integer"
+          },
           "resource" : {
             "type" : "object",
             "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:ResourceProfileInfo"

flink-runtime-web/src/test/resources/rest_api_v1.snapshot

@@ -4190,6 +4190,9 @@
               "jobId" : {
                 "type" : "any"
               },
+              "numberOfTasks" : {
+                "type" : "integer"
+              },
               "resource" : {
                 "type" : "object",
                 "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:ResourceProfileInfo"

flink-runtime-web/web-dashboard/src/app/interfaces/task-manager.ts

@@ -38,6 +38,7 @@ export interface TaskManagerDetail {
 
 export interface AllocatedSlot {
   jobId: string;
+  numberOfTasks: number;
   resource: Resources;
 }
 

flink-runtime-web/web-dashboard/src/app/pages/task-manager/metrics/task-manager-metrics.component.html

@@ -396,6 +396,7 @@
       <tr>
         <th>#</th>
         <th>Job ID</th>
+        <th>Tasks</th>
         <th>CPU (cores)</th>
         <th>Task Heap memory (MB)</th>
         <th>Task Off-Heap memory (MB)</th>
@@ -409,6 +410,7 @@
           <strong>{{ i | number }}</strong>
         </td>
         <td>{{ slot.jobId }}</td>
+        <td>{{ slot.numberOfTasks }}</td>
         <td>{{ slot.resource.cpuCores | number }}</td>
         <td>{{ slot.resource.taskHeapMemory | number }}</td>
         <td>{{ slot.resource.taskOffHeapMemory | number }}</td>

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DefaultSlotStatusSyncer.java

@@ -41,6 +41,7 @@
 
 import java.time.Duration;
 import java.util.HashSet;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -295,16 +296,15 @@ public boolean reportSlotStatus(InstanceID instanceId, SlotReport slotReport) {
     private boolean syncAllocatedSlotStatus(SlotStatus slotStatus, TaskManagerInfo taskManager) {
         final AllocationID allocationId = Preconditions.checkNotNull(slotStatus.getAllocationID());
         final JobID jobId = Preconditions.checkNotNull(slotStatus.getJobID());
+        final int numberOfTasks = slotStatus.getNumberOfTasks();
         try (MdcUtils.MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobId))) {
             final ResourceProfile resourceProfile =
                     Preconditions.checkNotNull(slotStatus.getResourceProfile());
 
-            if (taskManager.getAllocatedSlots().containsKey(allocationId)) {
-                if (taskManager.getAllocatedSlots().get(allocationId).getState()
-                        == SlotState.PENDING) {
+            TaskManagerSlotInformation slot = taskManager.getAllocatedSlots().get(allocationId);
+            if (Objects.nonNull(slot)) {
+                if (slot.getState() == SlotState.PENDING) {
                     // Allocation Complete
-                    final TaskManagerSlotInformation slot =
-                            taskManager.getAllocatedSlots().get(allocationId);
                     pendingSlotAllocations.remove(slot.getAllocationId());
                     taskManagerTracker.notifySlotStatus(
                             slot.getAllocationId(),
@@ -313,6 +313,7 @@ private boolean syncAllocatedSlotStatus(SlotStatus slotStatus, TaskManagerInfo t
                             slot.getResourceProfile(),
                             SlotState.ALLOCATED);
                 }
+                slot.setNumberOfTasks(numberOfTasks);
                 return true;
             } else {
                 Preconditions.checkState(
@@ -325,6 +326,7 @@ private boolean syncAllocatedSlotStatus(SlotStatus slotStatus, TaskManagerInfo t
                         resourceProfile,
                         SlotState.ALLOCATED);
                 resourceTracker.notifyAcquiredResource(jobId, resourceProfile);
+                taskManager.getAllocatedSlots().get(allocationId).setNumberOfTasks(numberOfTasks);
                 return false;
             }
         }

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java

@@ -803,7 +803,12 @@ public Collection<SlotInfo> getAllocatedSlotsOf(InstanceID instanceID) {
                 .map(Map::values)
                 .orElse(Collections.emptyList())
                 .stream()
-                .map(slot -> new SlotInfo(slot.getJobId(), slot.getResourceProfile()))
+                .map(
+                        slot ->
+                                new SlotInfo(
+                                        slot.getJobId(),
+                                        slot.getResourceProfile(),
+                                        slot.getNumberOfTasks()))
                 .collect(Collectors.toList());
     }
 

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerSlot.java

@@ -26,6 +26,8 @@
 import org.apache.flink.runtime.resourcemanager.registration.TaskExecutorConnection;
 import org.apache.flink.util.Preconditions;
 
+import javax.annotation.Nullable;
+
 import static org.apache.flink.util.Preconditions.checkArgument;
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -51,6 +53,8 @@ public class FineGrainedTaskManagerSlot implements TaskManagerSlotInformation {
     /** Current state of this slot. Should be either PENDING or ALLOCATED. */
     private SlotState state;
 
+    private @Nullable Integer numberOfTasks = null;
+
     public FineGrainedTaskManagerSlot(
             AllocationID allocationId,
             JobID jobId,
@@ -72,6 +76,17 @@ public ResourceProfile getResourceProfile() {
         return resourceProfile;
     }
 
+    @Override
+    public void setNumberOfTasks(int numberOfTasks) {
+        this.numberOfTasks = numberOfTasks;
+    }
+
+    @Nullable
+    @Override
+    public Integer getNumberOfTasks() {
+        return numberOfTasks;
+    }
+
     @Override
     public SlotState getState() {
         return state;

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/TaskManagerSlotInformation.java

@@ -60,4 +60,19 @@ default boolean isMatchingRequirement(ResourceProfile required) {
      * @return resource profile of this slot
      */
     ResourceProfile getResourceProfile();
+
+    /**
+     * Set the number of tasks that locate at the current slot.
+     *
+     * @param numberOfTasks The number of tasks at the current slot.
+     */
+    void setNumberOfTasks(int numberOfTasks);
+
+    /**
+     * Get the number of tasks that locate at the current slot.
+     *
+     * @return The number of tasks that locate at the current slot.
+     */
+    @Nullable
+    Integer getNumberOfTasks();
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/SlotInfo.java

@@ -32,6 +32,8 @@
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonSerialize;
 
+import javax.annotation.Nullable;
+
 import java.io.Serializable;
 import java.util.Objects;
 
@@ -46,31 +48,44 @@ public class SlotInfo implements ResponseBody, Serializable {
 
     public static final String FIELD_NAME_JOB_ID = "jobId";
 
+    public static final String FIELD_NAME_NUMBER_OF_TASKS = "numberOfTasks";
+
     @JsonProperty(FIELD_NAME_RESOURCE)
     private final ResourceProfileInfo resource;
 
     @JsonProperty(FIELD_NAME_JOB_ID)
     @JsonSerialize(using = JobIDSerializer.class)
     private final JobID jobId;
 
+    @JsonProperty(FIELD_NAME_NUMBER_OF_TASKS)
+    private final @Nullable Integer numberOfTasks;
+
     @JsonCreator
     public SlotInfo(
             @JsonDeserialize(using = JobIDDeserializer.class) @JsonProperty(FIELD_NAME_JOB_ID)
                     JobID jobId,
-            @JsonProperty(FIELD_NAME_RESOURCE) ResourceProfileInfo resource) {
+            @JsonProperty(FIELD_NAME_RESOURCE) ResourceProfileInfo resource,
+            @Nullable @JsonProperty(FIELD_NAME_NUMBER_OF_TASKS) Integer numberOfTasks) {
         this.jobId = Preconditions.checkNotNull(jobId);
         this.resource = Preconditions.checkNotNull(resource);
+        this.numberOfTasks = numberOfTasks;
     }
 
-    public SlotInfo(JobID jobId, ResourceProfile resource) {
-        this(jobId, ResourceProfileInfo.fromResourceProfile(resource));
+    public SlotInfo(JobID jobId, ResourceProfile resource, @Nullable Integer numberOfTasks) {
+        this(jobId, ResourceProfileInfo.fromResourceProfile(resource), numberOfTasks);
     }
 
     @JsonIgnore
     public JobID getJobId() {
         return jobId;
     }
 
+    @Nullable
+    @JsonIgnore
+    public Integer getNumberOfTasks() {
+        return numberOfTasks;
+    }
+
     @JsonIgnore
     public ResourceProfileInfo getResource() {
         return resource;
@@ -85,11 +100,13 @@ public boolean equals(Object o) {
             return false;
         }
         SlotInfo that = (SlotInfo) o;
-        return Objects.equals(jobId, that.jobId) && Objects.equals(resource, that.resource);
+        return Objects.equals(jobId, that.jobId)
+                && Objects.equals(resource, that.resource)
+                && Objects.equals(numberOfTasks, that.numberOfTasks);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(jobId, resource);
+        return Objects.hash(jobId, resource, numberOfTasks);
     }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotStatus.java

@@ -18,6 +18,7 @@
 
 package org.apache.flink.runtime.taskexecutor;
 
+import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -49,19 +50,33 @@ public class SlotStatus implements Serializable {
      */
     private final JobID jobID;
 
+    private final int numberOfTasks;
+
+    @VisibleForTesting
     public SlotStatus(SlotID slotID, ResourceProfile resourceProfile) {
-        this(slotID, resourceProfile, null, null);
+        this(slotID, resourceProfile, null, null, 0);
     }
 
+    @VisibleForTesting
     public SlotStatus(
             SlotID slotID,
             ResourceProfile resourceProfile,
             JobID jobID,
             AllocationID allocationID) {
+        this(slotID, resourceProfile, jobID, allocationID, 0);
+    }
+
+    public SlotStatus(
+            SlotID slotID,
+            ResourceProfile resourceProfile,
+            JobID jobID,
+            AllocationID allocationID,
+            int numberOfTasks) {
         this.slotID = checkNotNull(slotID, "slotID cannot be null");
         this.resourceProfile = checkNotNull(resourceProfile, "profile cannot be null");
         this.allocationID = allocationID;
         this.jobID = jobID;
+        this.numberOfTasks = numberOfTasks;
     }
 
     /**
@@ -100,6 +115,10 @@ public JobID getJobID() {
         return jobID;
     }
 
+    public int getNumberOfTasks() {
+        return numberOfTasks;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -143,6 +162,8 @@ public String toString() {
                 + allocationID
                 + ", jobID="
                 + jobID
+                + ", numberOfTasks="
+                + numberOfTasks
                 + ", resourceProfile="
                 + resourceProfile
                 + '}';