Addressed comments

Author: Minni Mittal

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ApplicationMasterService.java

@@ -398,9 +398,13 @@ public AllocateResponse allocate(AllocateRequest request)
         amrmTokenIdentifier.getApplicationAttemptId();
     RMAppAttemptMetrics rmMetrics = getAppAttemptMetrics(appAttemptId);
     // we do this here to prevent the internal lock in allocate()
-    rmMetrics.setAllocateLatenciesTimestamps(request.getAskList());
+    if (rmMetrics != null) {
+      rmMetrics.setAllocateLatenciesTimestamps(request.getAskList());
+    }
     AllocateResponse response = allocate(request, amrmTokenIdentifier);
-    rmMetrics.updateAllocateLatencies(response.getAllocatedContainers());
+    if (rmMetrics != null) {
+      rmMetrics.updateAllocateLatencies(response.getAllocatedContainers());
+    }
     return response;
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClusterMetrics.java

@@ -77,10 +77,10 @@ public class ClusterMetrics {
     MutableGaugeInt rmDispatcherEventQueueSize;
   @Metric("# of scheduler dispatcher event queue size")
     MutableGaugeInt schedulerDispatcherEventQueueSize;
-  @Metric("Allocation Latencies for Guarantee containers")
-  MutableQuantiles allocateLatencyGuarQuantiles;
+  @Metric("Allocation Latencies for Guaranteed containers")
+    MutableQuantiles allocateLatencyGuarQuantiles;
   @Metric("Allocation Latencies for Opportunistic containers")
-  MutableQuantiles allocateLatencyOppQuantiles;
+    MutableQuantiles allocateLatencyOppQuantiles;
   private boolean rmEventProcMonitorEnable = false;
 
   private static final MetricsInfo RECORD_INFO = info("ClusterMetrics",
@@ -130,14 +130,14 @@ public static ClusterMetrics getMetrics() {
   }
 
   private void initialize() {
-    allocateLatencyGuarQuantiles = registry
-        .newQuantiles("AllocateLatencyGuaranteed",
-            "Latency to fulfill an Allocate(Guaranteed) requests", "ops",
-            "latency", 5);
-    allocateLatencyOppQuantiles = registry
-        .newQuantiles("AllocateLatencyOpportunistic",
-            "Latency to fulfill an Allocate(Opportunistic) requests", "ops",
-            "latency", 5);
+    allocateLatencyGuarQuantiles = registry.newQuantiles(
+        "AllocateLatencyGuaranteed",
+        "Latency to fulfill an Allocate(Guaranteed) requests", "ops",
+        "latency", 5);
+    allocateLatencyOppQuantiles = registry.newQuantiles(
+        "AllocateLatencyOpportunistic",
+        "Latency to fulfill an Allocate(Opportunistic) requests", "ops",
+        "latency", 5);
   }
 
   private static void registerMetrics() {
@@ -400,4 +400,22 @@ public int getSchedulerEventQueueSize() {
   public void setSchedulerEventQueueSize(int schedulerEventQueueSize) {
     this.schedulerDispatcherEventQueueSize.set(schedulerEventQueueSize);
   }
+
+  public MutableQuantiles getAllocateLatencyGuarQuantiles() {
+    return allocateLatencyGuarQuantiles;
+  }
+
+  public void setAllocateLatencyGuarQuantiles(
+      MutableQuantiles allocateLatencyGuarQuantiles) {
+    this.allocateLatencyGuarQuantiles = allocateLatencyGuarQuantiles;
+  }
+
+  public MutableQuantiles getAllocateLatencyOppQuantiles() {
+    return allocateLatencyOppQuantiles;
+  }
+
+  public void setAllocateLatencyOppQuantiles(
+      MutableQuantiles allocateLatencyOppQuantiles) {
+    this.allocateLatencyOppQuantiles = allocateLatencyOppQuantiles;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/rmapp/attempt/RMAppAttemptMetrics.java

@@ -33,6 +33,7 @@
 import org.apache.hadoop.util.Time;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ExecutionType;
+import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.ClusterMetrics;
 import org.slf4j.Logger;
@@ -69,8 +70,10 @@ public class RMAppAttemptMetrics {
       new int[NodeType.values().length][NodeType.values().length];
   private volatile int totalAllocatedContainers;
 
-  private ConcurrentHashMap<Long, Long> allocationGuaranteedLatencies = null;
-  private ConcurrentHashMap<Long, Long> allocationOpportunisticLatencies = null;
+  private ConcurrentHashMap<Long, Long> allocationGuaranteedLatencies =
+      new ConcurrentHashMap<Long, Long>();
+  private ConcurrentHashMap<Long, Long> allocationOpportunisticLatencies =
+      new ConcurrentHashMap<Long, Long>();
 
   public RMAppAttemptMetrics(ApplicationAttemptId attemptId,
       RMContext rmContext) {
@@ -79,8 +82,6 @@ public RMAppAttemptMetrics(ApplicationAttemptId attemptId,
     this.readLock = lock.readLock();
     this.writeLock = lock.writeLock();
     this.rmContext = rmContext;
-    this.allocationGuaranteedLatencies = new ConcurrentHashMap<Long, Long>();
-    this.allocationOpportunisticLatencies = new ConcurrentHashMap<Long, Long>();
   }
 
   public void updatePreemptionInfo(Resource resource, RMContainer container) {
@@ -254,20 +255,9 @@ public void setApplicationAttemptHeadRoom(Resource headRoom) {
     this.applicationHeadroom = headRoom;
   }
 
-  /**
-   * Add allocationID latency to the application ID with a timestap =
-   * CurrentTime (guaranteed)
-   *
-   * @param allocId the allocation Id to add If the allocation ID is already
-   *                present (which shouldn't happen) it ignores the entry
-   */
-  public void addAllocationGuarLatencyIfNotExists(long allocId) {
-    addAllocationGuarLatencyIfNotExists(allocId, System.currentTimeMillis());
-  }
-
   /**
    * Add allocationID latency to the application ID with a specific timestamp
-   * (guaranteed)
+   * (guaranteed).
    *
    * @param allocId   allocationId
    * @param timestamp the timestamp to associate
@@ -277,21 +267,9 @@ public void addAllocationGuarLatencyIfNotExists(long allocId,
     allocationGuaranteedLatencies.putIfAbsent(allocId, timestamp);
   }
 
-  /**
-   * Add allocationID latency to the application ID with a timestap =
-   * CurrentTime (opportunistic)
-   *
-   * @param allocId the allocation Id to add If the allocation ID is already
-   *                present (which shouldn't happen) it ignores the entry
-   */
-  public void addAllocationOppLatencyIfNotExists(long allocId) {
-    this.addAllocationOppLatencyIfNotExists(allocId,
-        System.currentTimeMillis());
-  }
-
   /**
    * Add allocationID latency to the application ID with a specific timestamp
-   * (opportunistic)
+   * (opportunistic).
    *
    * @param allocId   allocationId
    * @param timestamp the timestamp to associate
@@ -301,36 +279,35 @@ public void addAllocationOppLatencyIfNotExists(long allocId, long timestamp) {
   }
 
   /**
-   * Returns the time associated when the allocation Id was added This method
-   * removes the allocation Id from the class (guaranteed)
+   * Returns the time associated when the allocation Id was added. This method
+   * removes the allocation Id from the class (guaranteed).
    *
    * @param allocId the allocation ID to get the associated time
    * @return the timestamp associated with that allocation id as well as stop
    * tracking it
    */
   public long getAndRemoveGuaAllocationLatencies(long allocId) {
-    Long ret = allocationGuaranteedLatencies.remove(new Long(allocId));
-    return ret != null ? ret : 0l;
+    Long ret = allocationGuaranteedLatencies.remove(allocId);
+    return ret != null ? ret : 0L;
   }
 
   /**
-   * Returns the time associated when the allocation Id was added This method
-   * removes the allocation Id from the class (opportunistic)
+   * Returns the time associated when the allocation Id was added. This method
+   * removes the allocation Id from the class (opportunistic).
    *
    * @param allocId the allocation ID to get the associated time
    * @return the timestamp associated with that allocation id as well as stop
    * tracking it
    */
   public long getAndRemoveOppAllocationLatencies(long allocId) {
-    Long ret = allocationOpportunisticLatencies.remove(new Long(allocId));
-    return ret != null ? ret : 0l;
+    Long ret = allocationOpportunisticLatencies.remove(allocId);
+    return ret != null ? ret : 0L;
   }
 
   /**
    * Set timestamp for the provided ResourceRequest. It will correctly identify
    * their ExecutionType, provided they have they have allocateId != 0 (DEFAULT)
-   * This is used in conjunction with This is used in conjunction with
-   * updatePromoteLatencies method method
+   * This is used in conjunction with updatePromoteLatencies method.
    *
    * @param requests the ResourceRequests to add.
    */
@@ -341,19 +318,18 @@ public void setAllocateLatenciesTimestamps(List<ResourceRequest> requests) {
         // we dont support tracking with negative or zero allocationIds
         long allocationRequestId = req.getAllocationRequestId();
         if (allocationRequestId > 0) {
-          if (req.getExecutionTypeRequest() != null) {
-            if (ExecutionType.GUARANTEED
-                .equals(req.getExecutionTypeRequest().getExecutionType())) {
+          ExecutionTypeRequest execReq = req.getExecutionTypeRequest();
+          if (execReq != null) {
+            if (ExecutionType.GUARANTEED.equals(execReq.getExecutionType())) {
               addAllocationGuarLatencyIfNotExists(allocationRequestId, now);
             } else {
               addAllocationOppLatencyIfNotExists(allocationRequestId, now);
             }
           }
         } else {
-          LOG.warn(String.format(
-              "Can't register allocate latency for %s container "
-                  + "with negative or zero allocation IDs",
-              req.getExecutionTypeRequest().getExecutionType()));
+          LOG.warn("Can't register allocate latency for {} container with"
+                  + "less than or equal to 0 allocation IDs",
+              req.getExecutionTypeRequest().getExecutionType());
         }
       }
     }
@@ -362,39 +338,36 @@ public void setAllocateLatenciesTimestamps(List<ResourceRequest> requests) {
   /**
    * Updated the JMX metrics class (ClusterMetrics) with the delta time when
    * these containers where added. It will correctly identify their
-   * ExecutionType, provided they have they have allocateId != 0 (DEFAULT)
+   * ExecutionType, provided they have they have allocateId != 0 (DEFAULT).
    *
    * @param response the list of the containers to allocate.
    */
   public void updateAllocateLatencies(List<Container> response) {
-
     for (Container container : response) {
       long allocationRequestId = container.getAllocationRequestId();
+      ExecutionType executionType = container.getExecutionType();
       // we dont support tracking with negative or zero allocationIds
       if (allocationRequestId > 0) {
         long now = System.currentTimeMillis();
-        long allocIdTime =
-            (container.getExecutionType() == ExecutionType.GUARANTEED) ?
-                getAndRemoveGuaAllocationLatencies(allocationRequestId) :
-                getAndRemoveOppAllocationLatencies(allocationRequestId);
+        long allocIdTime = (executionType == ExecutionType.GUARANTEED) ?
+            getAndRemoveGuaAllocationLatencies(allocationRequestId) :
+            getAndRemoveOppAllocationLatencies(allocationRequestId);
         if (allocIdTime != 0) {
-          if (container.getExecutionType() == ExecutionType.GUARANTEED) {
+          if (executionType == ExecutionType.GUARANTEED) {
             ClusterMetrics.getMetrics()
                 .addAllocateGuarLatencyEntry(now - allocIdTime);
           } else {
             ClusterMetrics.getMetrics()
                 .addAllocateOppLatencyEntry(now - allocIdTime);
           }
         } else {
-          LOG.error(String.format(
-              "Can't register allocate latency for %s container %s; allotTime=%d ",
-              container.getExecutionType(), container.getId(), allocIdTime));
+          LOG.error("Can't register allocate latency for {} container {}; "
+              + "allotTime={}", executionType, container.getId(), allocIdTime);
         }
       } else {
-        LOG.warn(String.format("Cant register promotion latency "
-                + "for container %s. Either allocationID is less than equal to 0 or "
-                + "lost the container ID", container.getExecutionType().name(),
-            container.getId()));
+        LOG.warn("Cant register promotion latency for {} container {}. Either "
+                + "allocationID is less than or equal to 0 or container is lost",
+            executionType, container.getId());
       }
     }
   }