[flink] branch master updated: [FLINK-23409][runtime][test] Add logs for FineGrainedSlotManager#checkResourceRequirements

[xtsong]

This is an automated email from the ASF dual-hosted git repository.

xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new b3eea98  [FLINK-23409][runtime][test] Add logs for 
FineGrainedSlotManager#checkResourceRequirements
b3eea98 is described below

commit b3eea9821aead6b8e11f592361384bed0a4fb016
Author: Yangze Guo <karma...@gmail.com>
AuthorDate: Mon Jul 19 10:45:04 2021 +0800

    [FLINK-23409][runtime][test] Add logs for 
FineGrainedSlotManager#checkResourceRequirements
    
    This closes #16530
---
 .../runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java   | 4 ++++
 1 file changed, 4 insertions(+)

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
index c570643..02a3f1c 100644
--- 
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
@@ -490,6 +490,7 @@ public class FineGrainedSlotManager implements SlotManager {
      */
     private void checkResourceRequirementsWithDelay() {
         if (lastResourceRequirementsCheck == null || 
lastResourceRequirementsCheck.isDone()) {
+            LOG.info("Scheduling the resource requirement check.");
             lastResourceRequirementsCheck =
                     scheduledExecutor.schedule(
                             () -> 
mainThreadExecutor.execute(this::checkResourceRequirements),
@@ -502,6 +503,7 @@ public class FineGrainedSlotManager implements SlotManager {
      * DO NOT call this method directly. Use {@link 
#checkResourceRequirementsWithDelay()} instead.
      */
     private void checkResourceRequirements() {
+        LOG.info("Matching resource requirements against available 
resources.");
         Map<JobID, Collection<ResourceRequirement>> missingResources =
                 resourceTracker.getMissingResources();
         if (missingResources.isEmpty()) {
@@ -519,9 +521,11 @@ public class FineGrainedSlotManager implements SlotManager 
{
                         missingResources, taskManagerTracker);
 
         // Allocate slots according to the result
+        LOG.info("Allocating slots for the resource requirements.");
         allocateSlotsAccordingTo(result.getAllocationsOnRegisteredResources());
 
         // Allocate task managers according to the result
+        LOG.info("Allocating task managers for the resource requirements.");
         final Set<PendingTaskManagerId> failAllocations =
                 
allocateTaskManagersAccordingTo(result.getPendingTaskManagersToAllocate());