[FLINK-38719][runtime] Display the number of tasks on the TaskManagers page.

RocMarshal · RocMarshal · commit efbbb4e390ef · 2025-12-29T16:23:52.000+08:00
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -5240,6 +5240,9 @@
           "freeSlots" : {
             "type" : "integer"
           },
+          "assignedTasks" : {
+            "type" : "integer"
+          },
           "hardware" : {
             "type" : "object",
             "id" : "urn:jsonschema:org:apache:flink:runtime:instance:HardwareDescription",
@@ -5464,6 +5467,9 @@
     "freeSlots" : {
       "type" : "integer"
     },
+    "assignedTasks" : {
+      "type" : "integer"
+    },
     "hardware" : {
       "type" : "object",
       "id" : "urn:jsonschema:org:apache:flink:runtime:instance:HardwareDescription",
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -3931,6 +3931,9 @@
               "freeSlots" : {
                 "type" : "integer"
               },
+              "assignedTasks" : {
+                "type" : "integer"
+              },
               "totalResource" : {
                 "type" : "object",
                 "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:ResourceProfileInfo",
@@ -4093,6 +4096,9 @@
         "freeSlots" : {
           "type" : "integer"
         },
+        "assignedTasks" : {
+          "type" : "integer"
+        },
         "totalResource" : {
           "type" : "object",
           "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:ResourceProfileInfo",
diff --git a/flink-runtime-web/web-dashboard/src/app/interfaces/task-manager.ts b/flink-runtime-web/web-dashboard/src/app/interfaces/task-manager.ts
@@ -27,6 +27,7 @@ export interface TaskManagerDetail {
   timeSinceLastHeartbeat: number;
   slotsNumber: number;
   freeSlots: number;
+  assignedTasks: number;
   hardware: Hardware;
   metrics: Metrics;
   memoryConfiguration: MemoryConfiguration;
@@ -68,6 +69,7 @@ export interface TaskManagersItem {
   timeSinceLastHeartbeat: number;
   slotsNumber: number;
   freeSlots: number;
+  assignedTasks: number;
   hardware: Hardware;
   blocked?: boolean;
 }
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/list/task-manager-list.component.html b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/list/task-manager-list.component.html
@@ -32,6 +32,7 @@
         <th [nzSortFn]="sortHeartBeatFn" [nzWidth]="'160px'">Last Heartbeat</th>
         <th [nzSortFn]="sortSlotsNumberFn" [nzWidth]="'90px'">All Slots</th>
         <th [nzSortFn]="sortFreeSlotsFn" [nzWidth]="'100px'">Free Slots</th>
+        <th [nzSortFn]="sortAssignedTasksFn" [nzWidth]="'100px'">Assigned Tasks</th>
         <th [nzSortFn]="sortCpuCoresFn" [nzWidth]="'110px'">CPU Cores</th>
         <th [nzSortFn]="sortPhysicalMemoryFn" [nzWidth]="'120px'">Physical MEM</th>
         <th [nzSortFn]="sortFreeMemoryFn" [nzWidth]="'130px'">JVM Heap Size</th>
@@ -53,6 +54,7 @@
           <td>{{ manager.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }}</td>
           <td>{{ manager.slotsNumber }}</td>
           <td>{{ manager.freeSlots }}</td>
+          <td>{{ manager.assignedTasks }}</td>
           <td>{{ manager.hardware.cpuCores }}</td>
           <td [attr.title]="manager.hardware.physicalMemory + ' bytes'">
             {{ manager.hardware.physicalMemory | humanizeBytes }}
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/list/task-manager-list.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/list/task-manager-list.component.ts
@@ -49,6 +49,7 @@ export class TaskManagerListComponent implements OnInit, OnDestroy {
   public readonly sortHeartBeatFn = createSortFn(item => item.timeSinceLastHeartbeat);
   public readonly sortSlotsNumberFn = createSortFn(item => item.slotsNumber);
   public readonly sortFreeSlotsFn = createSortFn(item => item.freeSlots);
+  public readonly sortAssignedTasksFn = createSortFn(item => item.assignedTasks);
   public readonly sortCpuCoresFn = createSortFn(item => item.hardware?.cpuCores);
   public readonly sortPhysicalMemoryFn = createSortFn(item => item.hardware?.physicalMemory);
   public readonly sortFreeMemoryFn = createSortFn(item => item.hardware?.freeMemory);
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html b/flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html
@@ -22,12 +22,15 @@
     <flink-blocked-badge *ngIf="taskManagerDetail?.blocked"></flink-blocked-badge>
   </div>
   <nz-descriptions *ngIf="taskManagerDetail" nzBordered nzSize="small">
-    <nz-descriptions-item [nzSpan]="2" nzTitle="Path">
+    <nz-descriptions-item [nzSpan]="1" nzTitle="Path">
       {{ taskManagerDetail.path }}
     </nz-descriptions-item>
     <nz-descriptions-item [nzSpan]="1" nzTitle="Free/All Slots">
       {{ taskManagerDetail.freeSlots }} / {{ taskManagerDetail.slotsNumber }}
     </nz-descriptions-item>
+    <nz-descriptions-item [nzSpan]="1" nzTitle="Assigned Tasks">
+      {{ taskManagerDetail.assignedTasks }}
+    </nz-descriptions-item>
     <nz-descriptions-item [nzSpan]="1" nzTitle="Last Heartbeat">
       {{ taskManagerDetail.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }}
     </nz-descriptions-item>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -687,6 +687,7 @@ public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Dur
                             taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId),
                             slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()),
                             slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()),
+                            slotManager.getAssignedTasksOf(taskExecutor.getInstanceID()),
                             slotManager.getRegisteredResourceOf(taskExecutor.getInstanceID()),
                             slotManager.getFreeResourceOf(taskExecutor.getInstanceID()),
                             taskExecutor.getHardwareDescription(),
@@ -717,6 +718,7 @@ public CompletableFuture<TaskManagerInfoWithSlots> requestTaskManagerDetailsInfo
                                     taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId),
                                     slotManager.getNumberRegisteredSlotsOf(instanceId),
                                     slotManager.getNumberFreeSlotsOf(instanceId),
+                                    slotManager.getAssignedTasksOf(instanceId),
                                     slotManager.getRegisteredResourceOf(instanceId),
                                     slotManager.getFreeResourceOf(instanceId),
                                     taskExecutor.getHardwareDescription(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/ClusterResourceStatisticsProvider.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/ClusterResourceStatisticsProvider.java
@@ -24,6 +24,9 @@
 /** Provides statistics of cluster resources. */
 public interface ClusterResourceStatisticsProvider {
 
+    /** Get total number of tasks assigned to the current instance. slots. */
+    int getAssignedTasks(InstanceID instanceId);
+
     /** Get total number of registered slots. */
     int getNumberRegisteredSlots();
 
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
@@ -755,6 +755,11 @@ private Set<PendingTaskManagerId> allocateTaskManagersAccordingTo(
     // Legacy APIs
     // ---------------------------------------------------------------------------------------------
 
+    @Override
+    public int getAssignedTasksOf(InstanceID instanceId) {
+        return taskManagerTracker.getAssignedTasks(instanceId);
+    }
+
     @Override
     public int getNumberRegisteredSlots() {
         return taskManagerTracker.getNumberRegisteredSlots();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTracker.java
@@ -35,6 +35,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -315,6 +316,22 @@ public Collection<PendingTaskManager> getPendingTaskManagers() {
                         Collections.emptySet()));
     }
 
+    @Override
+    public int getAssignedTasks(InstanceID instanceId) {
+        FineGrainedTaskManagerRegistration taskManagerRegistration =
+                taskManagerRegistrations.get(instanceId);
+        if (Objects.isNull(taskManagerRegistration)) {
+            return 0;
+        }
+        int totalAssignedTasks = 0;
+        for (TaskManagerSlotInformation slot :
+                taskManagerRegistration.getAllocatedSlots().values()) {
+            final int assignedTasks = slot.getAssignedTasks();
+            totalAssignedTasks = totalAssignedTasks + assignedTasks;
+        }
+        return totalAssignedTasks;
+    }
+
     @Override
     public int getNumberRegisteredSlots() {
         return taskManagerRegistrations.values().stream()
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
@@ -45,6 +45,9 @@
  * failure, respectively.
  */
 public interface SlotManager extends AutoCloseable {
+
+    int getAssignedTasksOf(InstanceID instanceId);
+
     int getNumberRegisteredSlots();
 
     int getNumberRegisteredSlotsOf(InstanceID instanceId);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerDetailsInfo.java
@@ -45,6 +45,8 @@ public class TaskManagerDetailsInfo extends TaskManagerInfo {
 
     public static final String FIELD_NAME_ALLOCATED_SLOTS = "allocatedSlots";
 
+    public static final String FIELD_NAME_ASSIGNED_TASKS = "assignedTasks";
+
     @JsonProperty(FIELD_NAME_METRICS)
     private final TaskManagerMetricsInfo taskManagerMetrics;
 
@@ -62,6 +64,7 @@ public TaskManagerDetailsInfo(
             @JsonProperty(FIELD_NAME_LAST_HEARTBEAT) long lastHeartbeat,
             @JsonProperty(FIELD_NAME_NUMBER_SLOTS) int numberSlots,
             @JsonProperty(FIELD_NAME_NUMBER_AVAILABLE_SLOTS) int numberAvailableSlots,
+            @JsonProperty(FIELD_NAME_ASSIGNED_TASKS) int assignedTasks,
             @JsonProperty(FIELD_NAME_TOTAL_RESOURCE) ResourceProfileInfo totalResource,
             @JsonProperty(FIELD_NAME_AVAILABLE_RESOURCE) ResourceProfileInfo freeResource,
             @JsonProperty(FIELD_NAME_HARDWARE) HardwareDescription hardwareDescription,
@@ -77,6 +80,7 @@ public TaskManagerDetailsInfo(
                 lastHeartbeat,
                 numberSlots,
                 numberAvailableSlots,
+                assignedTasks,
                 totalResource,
                 freeResource,
                 hardwareDescription,
@@ -98,6 +102,7 @@ public TaskManagerDetailsInfo(
                 taskManagerInfoWithSlots.getTaskManagerInfo().getLastHeartbeat(),
                 taskManagerInfoWithSlots.getTaskManagerInfo().getNumberSlots(),
                 taskManagerInfoWithSlots.getTaskManagerInfo().getNumberAvailableSlots(),
+                taskManagerInfoWithSlots.getTaskManagerInfo().getAssignedTasks(),
                 taskManagerInfoWithSlots.getTaskManagerInfo().getTotalResource(),
                 taskManagerInfoWithSlots.getTaskManagerInfo().getFreeResource(),
                 taskManagerInfoWithSlots.getTaskManagerInfo().getHardwareDescription(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfo.java
@@ -59,6 +59,8 @@ public class TaskManagerInfo implements ResponseBody, Serializable {
 
     public static final String FIELD_NAME_NUMBER_AVAILABLE_SLOTS = "freeSlots";
 
+    public static final String FIELD_NAME_ASSIGNED_TASKS = "assignedTasks";
+
     public static final String FIELD_NAME_TOTAL_RESOURCE = "totalResource";
 
     public static final String FIELD_NAME_AVAILABLE_RESOURCE = "freeResource";
@@ -93,6 +95,9 @@ public class TaskManagerInfo implements ResponseBody, Serializable {
     @JsonProperty(FIELD_NAME_NUMBER_AVAILABLE_SLOTS)
     private final int numberAvailableSlots;
 
+    @JsonProperty(FIELD_NAME_ASSIGNED_TASKS)
+    private final int assignedTasks;
+
     @JsonProperty(FIELD_NAME_TOTAL_RESOURCE)
     private final ResourceProfileInfo totalResource;
 
@@ -121,6 +126,7 @@ public TaskManagerInfo(
             @JsonProperty(FIELD_NAME_LAST_HEARTBEAT) long lastHeartbeat,
             @JsonProperty(FIELD_NAME_NUMBER_SLOTS) int numberSlots,
             @JsonProperty(FIELD_NAME_NUMBER_AVAILABLE_SLOTS) int numberAvailableSlots,
+            @JsonProperty(FIELD_NAME_ASSIGNED_TASKS) int assignedTasks,
             @JsonProperty(FIELD_NAME_TOTAL_RESOURCE) ResourceProfileInfo totalResource,
             @JsonProperty(FIELD_NAME_AVAILABLE_RESOURCE) ResourceProfileInfo freeResource,
             @JsonProperty(FIELD_NAME_HARDWARE) HardwareDescription hardwareDescription,
@@ -133,6 +139,7 @@ public TaskManagerInfo(
         this.lastHeartbeat = lastHeartbeat;
         this.numberSlots = numberSlots;
         this.numberAvailableSlots = numberAvailableSlots;
+        this.assignedTasks = assignedTasks;
         this.totalResource = totalResource;
         this.freeResource = freeResource;
         this.hardwareDescription = Preconditions.checkNotNull(hardwareDescription);
@@ -148,6 +155,7 @@ public TaskManagerInfo(
             long lastHeartbeat,
             int numberSlots,
             int numberAvailableSlots,
+            int assignedTasks,
             ResourceProfile totalResource,
             ResourceProfile freeResource,
             HardwareDescription hardwareDescription,
@@ -161,6 +169,7 @@ public TaskManagerInfo(
                 lastHeartbeat,
                 numberSlots,
                 numberAvailableSlots,
+                assignedTasks,
                 ResourceProfileInfo.fromResourceProfile(totalResource),
                 ResourceProfileInfo.fromResourceProfile(freeResource),
                 hardwareDescription,
@@ -203,6 +212,11 @@ public int getNumberAvailableSlots() {
         return numberAvailableSlots;
     }
 
+    @JsonIgnore
+    public int getAssignedTasks() {
+        return assignedTasks;
+    }
+
     @JsonIgnore
     public ResourceProfileInfo getTotalResource() {
         return totalResource;
@@ -242,6 +256,7 @@ public boolean equals(Object o) {
                 && lastHeartbeat == that.lastHeartbeat
                 && numberSlots == that.numberSlots
                 && numberAvailableSlots == that.numberAvailableSlots
+                && Objects.equals(assignedTasks, that.assignedTasks)
                 && Objects.equals(totalResource, that.totalResource)
                 && Objects.equals(freeResource, that.freeResource)
                 && Objects.equals(resourceId, that.resourceId)
@@ -261,6 +276,7 @@ public int hashCode() {
                 lastHeartbeat,
                 numberSlots,
                 numberAvailableSlots,
+                assignedTasks,
                 totalResource,
                 freeResource,
                 hardwareDescription,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTaskExecutorTest.java
@@ -432,6 +432,7 @@ void testRequestTaskManagerDetailsInfo() throws Exception {
             TaskManagerInfoWithSlots taskManagerInfoWithSlots =
                     rmGateway.requestTaskManagerDetailsInfo(taskExecutorResourceID, TIMEOUT).get();
 
+            assertThat(taskManagerInfoWithSlots.getTaskManagerInfo().getAssignedTasks()).isZero();
             assertThat(taskManagerInfoWithSlots.getAllocatedSlots())
                     .allMatch(slotInfo -> slotInfo.getAssignedTasks() == 0);
 
@@ -467,6 +468,7 @@ void testRequestTaskManagerDetailsInfo() throws Exception {
             // Test request task manager information after slot report.
             taskManagerInfoWithSlots =
                     rmGateway.requestTaskManagerDetailsInfo(taskExecutorResourceID, TIMEOUT).get();
+            assertThat(taskManagerInfoWithSlots.getTaskManagerInfo().getAssignedTasks()).isOne();
             assertThat(
                             taskManagerInfoWithSlots.getAllocatedSlots().stream()
                                     .map(SlotInfo::getAssignedTasks))
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
@@ -206,6 +206,7 @@ void testRequestTaskManagerInfo() throws Exception {
         assertThat(taskManagerInfo.getJmxPort()).isEqualTo(jmxPort);
         assertThat(taskManagerInfo.getNumberSlots()).isEqualTo(0);
         assertThat(taskManagerInfo.getNumberAvailableSlots()).isEqualTo(0);
+        assertThat(taskManagerInfo.getAssignedTasks()).isZero();
         assertThat(taskManagerInfoWithSlots.getAllocatedSlots()).isEmpty();
     }
 
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManager.java
@@ -54,6 +54,11 @@ public class TestingSlotManager implements SlotManager {
         this.triggerRequirementsCheckConsumer = triggerRequirementsCheckConsumer;
     }
 
+    @Override
+    public int getAssignedTasksOf(InstanceID instanceId) {
+        return 0;
+    }
+
     @Override
     public int getNumberRegisteredSlots() {
         return 0;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerDetailsHandlerTest.java
@@ -173,6 +173,7 @@ private static TaskManagerInfo createEmptyTaskManagerInfo() {
                 0L,
                 0,
                 0,
+                0,
                 ResourceProfile.ZERO,
                 ResourceProfile.ZERO,
                 new HardwareDescription(0, 0L, 0L, 0L),
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/taskmanager/TaskManagerInfoTest.java
@@ -55,6 +55,7 @@ static TaskManagerInfo createRandomTaskManagerInfo() {
                 random.nextLong(),
                 random.nextInt(),
                 random.nextInt(),
+                0,
                 ResourceProfile.ZERO,
                 ResourceProfile.ZERO,
                 new HardwareDescription(
