[FLINK-33391][runtime] Support tasks balancing at TM level for Adaptive Scheduler.

Author: RocMarshal

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

@@ -296,7 +296,7 @@ public Optional<ReservedSlots> tryReserveResources(JobSchedulingPlan jobScheduli
             final Map<ExecutionVertexID, LogicalSlot> assignedSlots = new HashMap<>();
 
             for (SlotAssignment assignment : jobSchedulingPlan.getSlotAssignments()) {
-                final SharedSlot sharedSlot = reserveSharedSlot(assignment.getSlotInfo());
+                final SharedSlot sharedSlot = reserveSharedSlot(assignment);
                 for (ExecutionVertexID executionVertexId :
                         assignment
                                 .getTargetAs(ExecutionSlotSharingGroup.class)
@@ -332,7 +332,10 @@ private boolean areAllExpectedSlotsAvailableAndFree(
         return true;
     }
 
-    private SharedSlot reserveSharedSlot(SlotInfo slotInfo) {
+    private SharedSlot reserveSharedSlot(SlotAssignment slotAssignment) {
+        final SlotInfo slotInfo = slotAssignment.getSlotInfo();
+        final ExecutionSlotSharingGroup requestedGroup =
+                slotAssignment.getTargetAs(ExecutionSlotSharingGroup.class);
         final PhysicalSlot physicalSlot =
                 reserveSlotFunction.reserveSlot(
                         slotInfo.getAllocationId(), ResourceProfile.UNKNOWN);