YARN-5938. Refactoring OpportunisticContainerAllocator to use SchedulerRequestKey instead of Priority and other misc fixes (asuresh)

(cherry picked from commit ac1e5d4f77)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/UpdateContainerRequest.java

@@ -158,6 +158,17 @@ public abstract class UpdateContainerRequest extends AbstractResourceRequest {
     return result;
   }
 
+  @Override
+  public String toString() {
+    return "UpdateReq{" +
+        "containerId=" + getContainerId() + ", " +
+        "containerVersion=" + getContainerVersion() + ", " +
+        "targetExecType=" + getExecutionType() + ", " +
+        "targetCapability=" + getCapability() + ", " +
+        "updateType=" + getContainerUpdateType() + ", " +
+        "}";
+  }
+
   @Override
   public boolean equals(Object obj) {
     if (this == obj) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/protocolrecords/impl/pb/AllocateResponsePBImpl.java

@@ -282,8 +282,8 @@ public class AllocateResponsePBImpl extends AllocateResponse {
       final List<Container> containers) {
     if (containers == null)
       return;
-    // this looks like a bug because it results in append and not set
     initLocalNewContainerList();
+    allocatedContainers.clear();
     allocatedContainers.addAll(containers);
   }
 
@@ -299,6 +299,7 @@ public class AllocateResponsePBImpl extends AllocateResponse {
     if (containers == null)
       return;
     initLocalUpdatedContainerList();
+    updatedContainers.clear();
     updatedContainers.addAll(containers);
   }
 
@@ -315,6 +316,7 @@ public class AllocateResponsePBImpl extends AllocateResponse {
     if (containers == null)
       return;
     initLocalFinishedContainerList();
+    completedContainersStatuses.clear();
     completedContainersStatuses.addAll(containers);
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api/protocolrecords/RemoteNode.java

@@ -87,4 +87,11 @@ public abstract class RemoteNode implements Comparable<RemoteNode> {
   public int compareTo(RemoteNode other) {
     return this.getNodeId().compareTo(other.getNodeId());
   }
+
+  @Override
+  public String toString() {
+    return "RemoteNode{" +
+        "nodeId=" + getNodeId() + ", " +
+        "httpAddress=" + getHttpAddress() + "}";
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerAllocator.java

@@ -22,8 +22,15 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.SecurityUtil;
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.*;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.api.records.ResourceBlacklistRequest;
+import org.apache.hadoop.yarn.api.records.ResourceRequest;
+import org.apache.hadoop.yarn.api.records.Token;
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.nodelabels.CommonNodeLabelsManager;
 import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
@@ -192,7 +199,8 @@ public class OpportunisticContainerAllocator {
 
   /**
    * Allocate OPPORTUNISTIC containers.
-   * @param request AllocateRequest
+   * @param blackList Resource BlackList Request
+   * @param oppResourceReqs Opportunistic Resource Requests
    * @param applicationAttemptId ApplicationAttemptId
    * @param opportContext App specific OpportunisticContainerContext
    * @param rmIdentifier RM Identifier
@@ -200,32 +208,24 @@ public class OpportunisticContainerAllocator {
    * @return List of Containers.
    * @throws YarnException YarnException
    */
-  public List<Container> allocateContainers(
+  public List<Container> allocateContainers(ResourceBlacklistRequest blackList,
-      AllocateRequest request, ApplicationAttemptId applicationAttemptId,
+      List<ResourceRequest> oppResourceReqs,
+      ApplicationAttemptId applicationAttemptId,
       OpportunisticContainerContext opportContext, long rmIdentifier,
       String appSubmitter) throws YarnException {
-    // Update released containers.
-    List<ContainerId> releasedContainers = request.getReleaseList();
-    int numReleasedContainers = releasedContainers.size();
-    if (numReleasedContainers > 0) {
-      LOG.info("AttemptID: " + applicationAttemptId + " released: "
-          + numReleasedContainers);
-      opportContext.getContainersAllocated().removeAll(releasedContainers);
-    }
 
     // Update black list.
-    ResourceBlacklistRequest rbr = request.getResourceBlacklistRequest();
+    if (blackList != null) {
-    if (rbr != null) {
+      opportContext.getBlacklist().removeAll(blackList.getBlacklistRemovals());
-      opportContext.getBlacklist().removeAll(rbr.getBlacklistRemovals());
+      opportContext.getBlacklist().addAll(blackList.getBlacklistAdditions());
-      opportContext.getBlacklist().addAll(rbr.getBlacklistAdditions());
     }
 
     // Add OPPORTUNISTIC requests to the outstanding ones.
-    opportContext.addToOutstandingReqs(request.getAskList());
+    opportContext.addToOutstandingReqs(oppResourceReqs);
 
     // Satisfy the outstanding OPPORTUNISTIC requests.
     List<Container> allocatedContainers = new ArrayList<>();
-    for (Priority priority :
+    for (SchedulerRequestKey schedulerKey :
         opportContext.getOutstandingOpReqs().descendingKeySet()) {
       // Allocated containers :
       //  Key = Requested Capability,
@@ -234,7 +234,7 @@ public class OpportunisticContainerAllocator {
       //          we need the requested capability (key) to match against
       //          the outstanding reqs)
       Map<Resource, List<Container>> allocated = allocate(rmIdentifier,
-          opportContext, priority, applicationAttemptId, appSubmitter);
+          opportContext, schedulerKey, applicationAttemptId, appSubmitter);
       for (Map.Entry<Resource, List<Container>> e : allocated.entrySet()) {
         opportContext.matchAllocationToOutstandingRequest(
             e.getKey(), e.getValue());
@@ -246,19 +246,22 @@ public class OpportunisticContainerAllocator {
   }
 
   private Map<Resource, List<Container>> allocate(long rmIdentifier,
-      OpportunisticContainerContext appContext, Priority priority,
+      OpportunisticContainerContext appContext, SchedulerRequestKey schedKey,
       ApplicationAttemptId appAttId, String userName) throws YarnException {
     Map<Resource, List<Container>> containers = new HashMap<>();
     for (ResourceRequest anyAsk :
-        appContext.getOutstandingOpReqs().get(priority).values()) {
+        appContext.getOutstandingOpReqs().get(schedKey).values()) {
       allocateContainersInternal(rmIdentifier, appContext.getAppParams(),
           appContext.getContainerIdGenerator(), appContext.getBlacklist(),
           appAttId, appContext.getNodeMap(), userName, containers, anyAsk);
+      if (!containers.isEmpty()) {
         LOG.info("Opportunistic allocation requested for ["
             + "priority=" + anyAsk.getPriority()
+            + ", allocationRequestId=" + anyAsk.getAllocationRequestId()
             + ", num_containers=" + anyAsk.getNumContainers()
             + ", capability=" + anyAsk.getCapability() + "]"
-          + " allocated = " + containers.get(anyAsk.getCapability()).size());
+            + " allocated = " + containers.keySet());
+      }
     }
     return containers;
   }
@@ -282,7 +285,9 @@ public class OpportunisticContainerAllocator {
       nodesForScheduling.add(nodeEntry.getValue());
     }
     if (nodesForScheduling.isEmpty()) {
-      LOG.warn("No nodes available for allocating opportunistic containers.");
+      LOG.warn("No nodes available for allocating opportunistic containers. [" +
+          "allNodes=" + allNodes + ", " +
+          "blacklist=" + blacklist + "]");
       return;
     }
     int numAllocated = 0;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/OpportunisticContainerContext.java

@@ -18,12 +18,7 @@
 
 package org.apache.hadoop.yarn.server.scheduler;
 
-import org.apache.hadoop.yarn.api.protocolrecords.AllocateResponse;
 import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
-import org.apache.hadoop.yarn.api.records.ExecutionType;
-import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.server.api.protocolrecords.RemoteNode;
@@ -52,9 +47,6 @@ public class OpportunisticContainerContext {
   private static final Logger LOG = LoggerFactory
       .getLogger(OpportunisticContainerContext.class);
 
-  // Currently just used to keep track of allocated containers.
-  // Can be used for reporting stats later.
-  private Set<ContainerId> containersAllocated = new HashSet<>();
   private AllocationParams appParams =
       new AllocationParams();
   private ContainerIdGenerator containerIdGenerator =
@@ -69,13 +61,9 @@ public class OpportunisticContainerContext {
   // Resource Name (host/rack/any) and capability. This mapping is required
   // to match a received Container to an outstanding OPPORTUNISTIC
   // ResourceRequest (ask).
-  private final TreeMap<Priority, Map<Resource, ResourceRequest>>
+  private final TreeMap<SchedulerRequestKey, Map<Resource, ResourceRequest>>
       outstandingOpReqs = new TreeMap<>();
 
-  public Set<ContainerId> getContainersAllocated() {
-    return containersAllocated;
-  }
-
   public AllocationParams getAppParams() {
     return appParams;
   }
@@ -119,20 +107,11 @@ public class OpportunisticContainerContext {
     return blacklist;
   }
 
-  public TreeMap<Priority, Map<Resource, ResourceRequest>>
+  public TreeMap<SchedulerRequestKey, Map<Resource, ResourceRequest>>
       getOutstandingOpReqs() {
     return outstandingOpReqs;
   }
 
-  public void updateCompletedContainers(AllocateResponse allocateResponse) {
-    for (ContainerStatus cs :
-        allocateResponse.getCompletedContainersStatuses()) {
-      if (cs.getExecutionType() == ExecutionType.OPPORTUNISTIC) {
-        containersAllocated.remove(cs.getContainerId());
-      }
-    }
-  }
-
   /**
    * Takes a list of ResourceRequests (asks), extracts the key information viz.
    * (Priority, ResourceName, Capability) and adds to the outstanding
@@ -144,7 +123,7 @@ public class OpportunisticContainerContext {
    */
   public void addToOutstandingReqs(List<ResourceRequest> resourceAsks) {
     for (ResourceRequest request : resourceAsks) {
-      Priority priority = request.getPriority();
+      SchedulerRequestKey schedulerKey = SchedulerRequestKey.create(request);
 
       // TODO: Extend for Node/Rack locality. We only handle ANY requests now
       if (!ResourceRequest.isAnyLocation(request.getResourceName())) {
@@ -156,10 +135,10 @@ public class OpportunisticContainerContext {
       }
 
       Map<Resource, ResourceRequest> reqMap =
-          outstandingOpReqs.get(priority);
+          outstandingOpReqs.get(schedulerKey);
       if (reqMap == null) {
         reqMap = new HashMap<>();
-        outstandingOpReqs.put(priority, reqMap);
+        outstandingOpReqs.put(schedulerKey, reqMap);
       }
 
       ResourceRequest resourceRequest = reqMap.get(request.getCapability());
@@ -171,7 +150,8 @@ public class OpportunisticContainerContext {
             resourceRequest.getNumContainers() + request.getNumContainers());
       }
       if (ResourceRequest.isAnyLocation(request.getResourceName())) {
-        LOG.info("# of outstandingOpReqs in ANY (at priority = " + priority
+        LOG.info("# of outstandingOpReqs in ANY (at" +
+            "priority = "+ schedulerKey.getPriority()
             + ", with capability = " + request.getCapability() + " ) : "
             + resourceRequest.getNumContainers());
       }
@@ -187,9 +167,9 @@ public class OpportunisticContainerContext {
   public void matchAllocationToOutstandingRequest(Resource capability,
       List<Container> allocatedContainers) {
     for (Container c : allocatedContainers) {
-      containersAllocated.add(c.getId());
+      SchedulerRequestKey schedulerKey = SchedulerRequestKey.extractFrom(c);
       Map<Resource, ResourceRequest> asks =
-          outstandingOpReqs.get(c.getPriority());
+          outstandingOpReqs.get(schedulerKey);
 
       if (asks == null) {
         continue;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/scheduler/SchedulerRequestKey.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.yarn.server.resourcemanager.scheduler;
+package org.apache.hadoop.yarn.server.scheduler;
 
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.Priority;
@@ -53,7 +53,7 @@ public final class SchedulerRequestKey implements
         container.getAllocationRequestId());
   }
 
-  private SchedulerRequestKey(Priority priority, long allocationRequestId) {
+  SchedulerRequestKey(Priority priority, long allocationRequestId) {
     this.priority = priority;
     this.allocationRequestId = allocationRequestId;
   }
@@ -119,4 +119,12 @@ public final class SchedulerRequestKey implements
         getAllocationRequestId() >>> 32));
     return result;
   }
+
+  @Override
+  public String toString() {
+    return "SchedulerRequestKey{" +
+        "priority=" + priority +
+        ", allocationRequestId=" + allocationRequestId +
+        '}';
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/scheduler/DistributedScheduler.java

@@ -227,10 +227,10 @@ public final class DistributedScheduler extends AbstractRequestInterceptor {
         .partitionAskList(request.getAllocateRequest().getAskList());
 
     // Allocate OPPORTUNISTIC containers.
-    request.getAllocateRequest().setAskList(partitionedAsks.getOpportunistic());
     List<Container> allocatedContainers =
         containerAllocator.allocateContainers(
-            request.getAllocateRequest(), applicationAttemptId,
+            request.getAllocateRequest().getResourceBlacklistRequest(),
+            partitionedAsks.getOpportunistic(), applicationAttemptId,
             oppContainerContext, rmIdentifier, appSubmitter);
 
     // Prepare request for sending to RM for scheduling GUARANTEED containers.
@@ -252,18 +252,11 @@ public final class DistributedScheduler extends AbstractRequestInterceptor {
       nodeTokens.put(nmToken.getNodeId(), nmToken);
     }
 
-    oppContainerContext.updateCompletedContainers(dsResp.getAllocateResponse());
-
     // Check if we have NM tokens for all the allocated containers. If not
     // generate one and update the response.
     updateAllocateResponse(
         dsResp.getAllocateResponse(), nmTokens, allocatedContainers);
 
-    if (LOG.isDebugEnabled()) {
-      LOG.debug("Number of opportunistic containers currently" +
-          "allocated by application: " + oppContainerContext
-          .getContainersAllocated().size());
-    }
     return dsResp;
   }
 }

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

@@ -400,7 +400,6 @@ public class ApplicationMasterService extends AbstractService implements
 
     ApplicationAttemptId appAttemptId =
         amrmTokenIdentifier.getApplicationAttemptId();
-    ApplicationId applicationId = appAttemptId.getApplicationId();
 
     this.amLivelinessMonitor.receivedPing(appAttemptId);
 
@@ -417,8 +416,10 @@ public class ApplicationMasterService extends AbstractService implements
       AllocateResponse lastResponse = lock.getAllocateResponse();
       if (!hasApplicationMasterRegistered(appAttemptId)) {
         String message =
-            "AM is not registered for known application attempt: " + appAttemptId
+            "AM is not registered for known application attempt: "
-                + " or RM had restarted after AM registered . AM should re-register.";
+                + appAttemptId
+                + " or RM had restarted after AM registered. "
+                + " AM should re-register.";
         throw new ApplicationMasterNotRegisteredException(message);
       }
 
@@ -433,12 +434,60 @@ public class ApplicationMasterService extends AbstractService implements
         throw new InvalidApplicationMasterRequestException(message);
       }
 
+      AllocateResponse response =
+          recordFactory.newRecordInstance(AllocateResponse.class);
+      allocateInternal(amrmTokenIdentifier.getApplicationAttemptId(),
+          request, response);
+
+      // update AMRMToken if the token is rolled-up
+      MasterKeyData nextMasterKey =
+          this.rmContext.getAMRMTokenSecretManager().getNextMasterKeyData();
+
+      if (nextMasterKey != null
+          && nextMasterKey.getMasterKey().getKeyId() != amrmTokenIdentifier
+          .getKeyId()) {
+        RMApp app =
+            this.rmContext.getRMApps().get(appAttemptId.getApplicationId());
+        RMAppAttempt appAttempt = app.getRMAppAttempt(appAttemptId);
+        RMAppAttemptImpl appAttemptImpl = (RMAppAttemptImpl)appAttempt;
+        Token<AMRMTokenIdentifier> amrmToken = appAttempt.getAMRMToken();
+        if (nextMasterKey.getMasterKey().getKeyId() !=
+            appAttemptImpl.getAMRMTokenKeyId()) {
+          LOG.info("The AMRMToken has been rolled-over. Send new AMRMToken back"
+              + " to application: " + appAttemptId.getApplicationId());
+          amrmToken = rmContext.getAMRMTokenSecretManager()
+              .createAndGetAMRMToken(appAttemptId);
+          appAttemptImpl.setAMRMToken(amrmToken);
+        }
+        response.setAMRMToken(org.apache.hadoop.yarn.api.records.Token
+            .newInstance(amrmToken.getIdentifier(), amrmToken.getKind()
+                .toString(), amrmToken.getPassword(), amrmToken.getService()
+                .toString()));
+      }
+
+      /*
+       * As we are updating the response inside the lock object so we don't
+       * need to worry about unregister call occurring in between (which
+       * removes the lock object).
+       */
+      response.setResponseId(lastResponse.getResponseId() + 1);
+      lock.setAllocateResponse(response);
+      return response;
+    }    
+  }
+
+  protected void allocateInternal(ApplicationAttemptId appAttemptId,
+      AllocateRequest request, AllocateResponse allocateResponse)
+      throws YarnException {
+
     //filter illegal progress values
     float filteredProgress = request.getProgress();
-      if (Float.isNaN(filteredProgress) || filteredProgress == Float.NEGATIVE_INFINITY
+    if (Float.isNaN(filteredProgress) ||
-        || filteredProgress < 0) {
+        filteredProgress == Float.NEGATIVE_INFINITY ||
+        filteredProgress < 0) {
       request.setProgress(0);
-      } else if (filteredProgress > 1 || filteredProgress == Float.POSITIVE_INFINITY) {
+    } else if (filteredProgress > 1 ||
+        filteredProgress == Float.POSITIVE_INFINITY) {
       request.setProgress(1);
     }
 
@@ -459,7 +508,7 @@ public class ApplicationMasterService extends AbstractService implements
         (blacklistRequest != null) ?
             blacklistRequest.getBlacklistRemovals() : Collections.EMPTY_LIST;
     RMApp app =
-          this.rmContext.getRMApps().get(applicationId);
+        this.rmContext.getRMApps().get(appAttemptId.getApplicationId());
 
     // set label expression for Resource Requests if resourceName=ANY
     ApplicationSubmissionContext asc = app.getApplicationSubmissionContext();
@@ -508,9 +557,9 @@ public class ApplicationMasterService extends AbstractService implements
     List<UpdateContainerRequest> increaseResourceReqs = new ArrayList<>();
     List<UpdateContainerRequest> decreaseResourceReqs = new ArrayList<>();
     List<UpdateContainerError> updateContainerErrors =
-          RMServerUtils.validateAndSplitUpdateResourceRequests(rmContext,
+        RMServerUtils.validateAndSplitUpdateResourceRequests(
-              request, maximumCapacity, increaseResourceReqs,
+            rmContext, request, maximumCapacity,
-              decreaseResourceReqs);
+            increaseResourceReqs, decreaseResourceReqs);
 
     // Send new requests to appAttempt.
     Allocation allocation;
@@ -535,17 +584,15 @@ public class ApplicationMasterService extends AbstractService implements
           "blacklistRemovals: " + blacklistRemovals);
     }
     RMAppAttempt appAttempt = app.getRMAppAttempt(appAttemptId);
-      AllocateResponse allocateResponse =
+
-          recordFactory.newRecordInstance(AllocateResponse.class);
     if (allocation.getNMTokens() != null &&
         !allocation.getNMTokens().isEmpty()) {
       allocateResponse.setNMTokens(allocation.getNMTokens());
     }
 
     // Notify the AM of container update errors
-      if (!updateContainerErrors.isEmpty()) {
+    addToUpdateContainerErrors(allocateResponse, updateContainerErrors);
-        allocateResponse.setUpdateErrors(updateContainerErrors);
+
-      }
     // update the response with the deltas of node status changes
     List<RMNode> updatedNodes = new ArrayList<RMNode>();
     if(app.pullRMNodeUpdates(updatedNodes) > 0) {
@@ -572,30 +619,21 @@ public class ApplicationMasterService extends AbstractService implements
       allocateResponse.setUpdatedNodes(updatedNodeReports);
     }
 
-      allocateResponse.setAllocatedContainers(allocation.getContainers());
+    addToAllocatedContainers(allocateResponse, allocation.getContainers());
+
     allocateResponse.setCompletedContainersStatuses(appAttempt
         .pullJustFinishedContainers());
-      allocateResponse.setResponseId(lastResponse.getResponseId() + 1);
     allocateResponse.setAvailableResources(allocation.getResourceLimit());
 
-      // Handling increased/decreased containers
+    // Handling increased containers
-      List<UpdatedContainer> updatedContainers = new ArrayList<>();
+    addToUpdatedContainers(
-      if (allocation.getIncreasedContainers() != null) {
+        allocateResponse, ContainerUpdateType.INCREASE_RESOURCE,
-        for (Container c : allocation.getIncreasedContainers()) {
+        allocation.getIncreasedContainers());
-          updatedContainers.add(
-              UpdatedContainer.newInstance(
-                  ContainerUpdateType.INCREASE_RESOURCE, c));
-        }
-      }
-      if (allocation.getDecreasedContainers() != null) {
-        for (Container c : allocation.getDecreasedContainers()) {
-          updatedContainers.add(
-              UpdatedContainer.newInstance(
-                  ContainerUpdateType.DECREASE_RESOURCE, c));
-        }
-      }
 
-      allocateResponse.setUpdatedContainers(updatedContainers);
+    // Handling decreased containers
+    addToUpdatedContainers(
+        allocateResponse, ContainerUpdateType.DECREASE_RESOURCE,
+        allocation.getDecreasedContainers());
 
     allocateResponse.setNumClusterNodes(this.rScheduler.getNumClusterNodes());
 
@@ -606,38 +644,44 @@ public class ApplicationMasterService extends AbstractService implements
     // Set application priority
     allocateResponse.setApplicationPriority(app
         .getApplicationPriority());
-
-      // update AMRMToken if the token is rolled-up
-      MasterKeyData nextMasterKey =
-          this.rmContext.getAMRMTokenSecretManager().getNextMasterKeyData();
-
-      if (nextMasterKey != null
-          && nextMasterKey.getMasterKey().getKeyId() != amrmTokenIdentifier
-            .getKeyId()) {
-        RMAppAttemptImpl appAttemptImpl = (RMAppAttemptImpl)appAttempt;
-        Token<AMRMTokenIdentifier> amrmToken = appAttempt.getAMRMToken();
-        if (nextMasterKey.getMasterKey().getKeyId() !=
-            appAttemptImpl.getAMRMTokenKeyId()) {
-          LOG.info("The AMRMToken has been rolled-over. Send new AMRMToken back"
-              + " to application: " + applicationId);
-          amrmToken = rmContext.getAMRMTokenSecretManager()
-              .createAndGetAMRMToken(appAttemptId);
-          appAttemptImpl.setAMRMToken(amrmToken);
-        }
-        allocateResponse.setAMRMToken(org.apache.hadoop.yarn.api.records.Token
-          .newInstance(amrmToken.getIdentifier(), amrmToken.getKind()
-            .toString(), amrmToken.getPassword(), amrmToken.getService()
-            .toString()));
   }
 
-      /*
+  protected void addToUpdateContainerErrors(AllocateResponse allocateResponse,
-       * As we are updating the response inside the lock object so we don't
+      List<UpdateContainerError> updateContainerErrors) {
-       * need to worry about unregister call occurring in between (which
+    if (!updateContainerErrors.isEmpty()) {
-       * removes the lock object).
+      if (allocateResponse.getUpdateErrors() != null
-       */
+          && !allocateResponse.getUpdateErrors().isEmpty()) {
-      lock.setAllocateResponse(allocateResponse);
+        updateContainerErrors = new ArrayList<>(updateContainerErrors);
-      return allocateResponse;
+        updateContainerErrors.addAll(allocateResponse.getUpdateErrors());
       }
+      allocateResponse.setUpdateErrors(updateContainerErrors);
+    }
+  }
+
+  protected void addToUpdatedContainers(AllocateResponse allocateResponse,
+      ContainerUpdateType updateType, List<Container> updatedContainers) {
+    if (updatedContainers != null && updatedContainers.size() > 0) {
+      ArrayList<UpdatedContainer> containersToSet = new ArrayList<>();
+      if (allocateResponse.getUpdatedContainers() != null &&
+          !allocateResponse.getUpdatedContainers().isEmpty()) {
+        containersToSet.addAll(allocateResponse.getUpdatedContainers());
+      }
+      for (Container updatedContainer : updatedContainers) {
+        containersToSet.add(
+            UpdatedContainer.newInstance(updateType, updatedContainer));
+      }
+      allocateResponse.setUpdatedContainers(containersToSet);
+    }
+  }
+
+  protected void addToAllocatedContainers(AllocateResponse allocateResponse,
+      List<Container> allocatedContainers) {
+    if (allocateResponse.getAllocatedContainers() != null
+        && !allocateResponse.getAllocatedContainers().isEmpty()) {
+      allocatedContainers = new ArrayList<>(allocatedContainers);
+      allocatedContainers.addAll(allocateResponse.getAllocatedContainers());
+    }
+    allocateResponse.setAllocatedContainers(allocatedContainers);
   }
 
   private PreemptionMessage generatePreemptionMessage(Allocation allocation){

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

@@ -101,8 +101,8 @@ public class OpportunisticContainerAllocatorAMService
   private final int k;
 
   private final long cacheRefreshInterval;
-  private List<RemoteNode> cachedNodes;
+  private volatile List<RemoteNode> cachedNodes;
-  private long lastCacheUpdateTime;
+  private volatile long lastCacheUpdateTime;
 
   public OpportunisticContainerAllocatorAMService(RMContext rmContext,
       YarnScheduler scheduler) {
@@ -218,8 +218,9 @@ public class OpportunisticContainerAllocatorAMService
   }
 
   @Override
-  public AllocateResponse allocate(AllocateRequest request) throws
+  protected void allocateInternal(ApplicationAttemptId appAttemptId,
-      YarnException, IOException {
+      AllocateRequest request, AllocateResponse allocateResponse)
+      throws YarnException {
 
     // Partition requests to GUARANTEED and OPPORTUNISTIC.
     OpportunisticContainerAllocator.PartitionedResourceRequests
@@ -227,40 +228,30 @@ public class OpportunisticContainerAllocatorAMService
         oppContainerAllocator.partitionAskList(request.getAskList());
 
     // Allocate OPPORTUNISTIC containers.
-    request.setAskList(partitionedAsks.getOpportunistic());
+    SchedulerApplicationAttempt appAttempt =
-    final ApplicationAttemptId appAttemptId = getAppAttemptId();
+        ((AbstractYarnScheduler)rmContext.getScheduler())
-    SchedulerApplicationAttempt appAttempt = ((AbstractYarnScheduler)
+            .getApplicationAttempt(appAttemptId);
-        rmContext.getScheduler()).getApplicationAttempt(appAttemptId);
 
     OpportunisticContainerContext oppCtx =
         appAttempt.getOpportunisticContainerContext();
     oppCtx.updateNodeList(getLeastLoadedNodes());
 
     List<Container> oppContainers =
-        oppContainerAllocator.allocateContainers(request, appAttemptId, oppCtx,
+        oppContainerAllocator.allocateContainers(
+            request.getResourceBlacklistRequest(),
+            partitionedAsks.getOpportunistic(), appAttemptId, oppCtx,
             ResourceManager.getClusterTimeStamp(), appAttempt.getUser());
 
     // Create RMContainers and update the NMTokens.
     if (!oppContainers.isEmpty()) {
       handleNewContainers(oppContainers, false);
       appAttempt.updateNMTokens(oppContainers);
+      addToAllocatedContainers(allocateResponse, oppContainers);
     }
 
     // Allocate GUARANTEED containers.
     request.setAskList(partitionedAsks.getGuaranteed());
-    AllocateResponse allocateResp = super.allocate(request);
+    super.allocateInternal(appAttemptId, request, allocateResponse);
-
-    // Add allocated OPPORTUNISTIC containers to the AllocateResponse.
-    if (!oppContainers.isEmpty()) {
-      allocateResp.getAllocatedContainers().addAll(oppContainers);
-    }
-
-    // Update opportunistic container context with the allocated GUARANTEED
-    // containers.
-    oppCtx.updateCompletedContainers(allocateResp);
-
-    // Add all opportunistic containers
-    return allocateResp;
   }
 
   @Override
@@ -387,11 +378,13 @@ public class OpportunisticContainerAllocatorAMService
   private synchronized List<RemoteNode> getLeastLoadedNodes() {
     long currTime = System.currentTimeMillis();
     if ((currTime - lastCacheUpdateTime > cacheRefreshInterval)
-        || cachedNodes == null) {
+        || (cachedNodes == null)) {
       cachedNodes = convertToRemoteNodes(
           this.nodeMonitor.selectLeastLoadedNodes(this.k));
+      if (cachedNodes.size() > 0) {
         lastCacheUpdateTime = currTime;
       }
+    }
     return cachedNodes;
   }
 

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

@@ -25,14 +25,13 @@ import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerReport;
 import org.apache.hadoop.yarn.api.records.ContainerState;
-import org.apache.hadoop.yarn.api.records.ContainerStatus;
 import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 
 /**

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

@@ -53,7 +53,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.event.RMAppAt
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanContainerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode
     .RMNodeDecreaseContainerEvent;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
 import org.apache.hadoop.yarn.state.MultipleArcTransition;
 import org.apache.hadoop.yarn.state.SingleArcTransition;

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

@@ -21,7 +21,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.rmcontainer;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 /**
  * The event signifying that a container has been reserved.

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

@@ -37,6 +37,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerStat
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.LocalitySchedulingPlacementSet;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.ResourceRequestUpdateResult;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
+
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.util.resource.Resources;
 
 import java.util.ArrayList;

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

@@ -51,7 +51,6 @@ import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.LogAggregationContext;
 import org.apache.hadoop.yarn.api.records.NMToken;
 import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.NodeLabel;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
@@ -73,12 +72,11 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerStat
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerUpdatesAcquiredEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeCleanContainerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.PlacementSet;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.SchedulableEntity;
 
 import org.apache.hadoop.yarn.server.scheduler.OpportunisticContainerContext;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
 import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.Resources;

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

@@ -40,6 +40,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsMana
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.util.resource.Resources;
 
 import com.google.common.collect.ImmutableSet;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java

@@ -34,6 +34,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerExitStatus;
 import org.apache.hadoop.yarn.api.records.ContainerStatus;
+import org.apache.hadoop.yarn.api.records.ExecutionType;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
@@ -2216,7 +2217,8 @@ public class LeafQueue extends AbstractCSQueue {
   @Override
   public void attachContainer(Resource clusterResource,
       FiCaSchedulerApp application, RMContainer rmContainer) {
-    if (application != null) {
+    if (application != null && rmContainer != null
+        && rmContainer.getExecutionType() == ExecutionType.GUARANTEED) {
       FiCaSchedulerNode node =
           scheduler.getNode(rmContainer.getContainer().getNodeId());
       allocateResource(clusterResource, application, rmContainer.getContainer()
@@ -2234,7 +2236,8 @@ public class LeafQueue extends AbstractCSQueue {
   @Override
   public void detachContainer(Resource clusterResource,
       FiCaSchedulerApp application, RMContainer rmContainer) {
-    if (application != null) {
+    if (application != null && rmContainer != null
+          && rmContainer.getExecutionType() == ExecutionType.GUARANTEED) {
       FiCaSchedulerNode node =
           scheduler.getNode(rmContainer.getContainer().getNodeId());
       releaseResource(clusterResource, application, rmContainer.getContainer()

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/IncreaseContainerAllocator.java

@@ -31,7 +31,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedContainerChangeRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSAssignment;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.SchedulingMode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/allocator/RegularContainerAllocator.java

@@ -38,7 +38,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppUtils;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
 
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivityDiagnosticConstant;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/SchedulerContainer.java

@@ -22,7 +22,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerState;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 /**
  * Contexts for a container inside scheduler

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerApp.java

@@ -52,7 +52,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedContainerChangeRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.activities.ActivitiesManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.AbstractCSQueue;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSAMContainerLaunchDiagnosticsConstants;
@@ -69,6 +68,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCo
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.SchedulerContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.PlacementSet;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement.SchedulingPlacementSet;
+
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.ResourceCalculator;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/common/fica/FiCaSchedulerNode.java

@@ -29,7 +29,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
 
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
 import org.apache.hadoop.yarn.util.resource.Resources;
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/distributed/NodeQueueLoadMonitor.java

@@ -30,7 +30,6 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
@@ -61,9 +60,9 @@ public class NodeQueueLoadMonitor implements ClusterMonitor {
     @Override
     public int compare(ClusterNode o1, ClusterNode o2) {
       if (getMetric(o1) == getMetric(o2)) {
-        return o1.timestamp < o2.timestamp ? +1 : -1;
+        return (int)(o2.timestamp - o1.timestamp);
       }
-      return getMetric(o1) > getMetric(o2) ? +1 : -1;
+      return getMetric(o1) - getMetric(o2);
     }
 
     public int getMetric(ClusterNode c) {
@@ -115,8 +114,13 @@ public class NodeQueueLoadMonitor implements ClusterMonitor {
       ReentrantReadWriteLock.WriteLock writeLock = sortedNodesLock.writeLock();
       writeLock.lock();
       try {
+        try {
+          List<NodeId> nodeIds = sortNodes();
           sortedNodes.clear();
-        sortedNodes.addAll(sortNodes());
+          sortedNodes.addAll(nodeIds);
+        } catch (Exception ex) {
+          LOG.warn("Got Exception while sorting nodes..", ex);
+        }
         if (thresholdCalculator != null) {
           thresholdCalculator.update();
         }
@@ -273,7 +277,7 @@ public class NodeQueueLoadMonitor implements ClusterMonitor {
       List<NodeId> retVal = ((k < this.sortedNodes.size()) && (k >= 0)) ?
           new ArrayList<>(this.sortedNodes).subList(0, k) :
           new ArrayList<>(this.sortedNodes);
-      return Collections.unmodifiableList(retVal);
+      return retVal;
     } finally {
       readLock.unlock();
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSAppAttempt.java

@@ -56,7 +56,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.Resources;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FSSchedulerNode.java

@@ -26,7 +26,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
 
 @Private

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoAppAttempt.java

@@ -33,10 +33,12 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ActiveUsersManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Queue;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
 
+
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
+
 import java.util.List;
 
 public class FifoAppAttempt extends FiCaSchedulerApp {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java

@@ -69,7 +69,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedContainerCha
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerAppUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplication;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAddedSchedulerEvent;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/LocalitySchedulingPlacementSet.java

@@ -25,7 +25,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.AppSchedulingInfo
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -177,11 +177,15 @@ public class LocalitySchedulingPlacementSet<N extends SchedulerNode>
         offSwitchRequest.getCapability());
   }
 
-  private ResourceRequest cloneResourceRequest(ResourceRequest request) {
+  public ResourceRequest cloneResourceRequest(ResourceRequest request) {
-    ResourceRequest newRequest =
+    ResourceRequest newRequest = ResourceRequest.newBuilder()
-        ResourceRequest.newInstance(request.getPriority(),
+        .priority(request.getPriority())
-            request.getResourceName(), request.getCapability(), 1,
+        .allocationRequestId(request.getAllocationRequestId())
-            request.getRelaxLocality(), request.getNodeLabelExpression());
+        .resourceName(request.getResourceName())
+        .capability(request.getCapability())
+        .numContainers(1)
+        .relaxLocality(request.getRelaxLocality())
+        .nodeLabelExpression(request.getNodeLabelExpression()).build();
     return newRequest;
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/placement/SchedulingPlacementSet.java

@@ -21,7 +21,7 @@ package org.apache.hadoop.yarn.server.resourcemanager.scheduler.placement;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 import java.util.Collection;
 import java.util.Iterator;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Application.java

@@ -61,7 +61,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/Task.java

@@ -31,7 +31,7 @@ import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.Priority;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
     .TestUtils;
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/ProportionalCapacityPreemptionPolicyMockFramework.java

@@ -36,8 +36,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
-    .SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicy.java

@@ -35,8 +35,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.monitor.SchedulingMonitor;
 import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
-    .SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestAppSchedulingInfo.java

@@ -33,6 +33,7 @@ import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSLeafQueue;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.junit.Assert;
 import org.junit.Test;
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/TestSchedulerApplicationAttempt.java

@@ -41,6 +41,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainerImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
 import org.junit.After;
 import org.junit.Test;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestCapacityScheduler.java

@@ -120,7 +120,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplication;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
 
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNodeReport;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.TestSchedulerUtils;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java

@@ -77,7 +77,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueStateManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 
 
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestReservations.java

@@ -60,7 +60,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceLimits;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceUsage;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt.AMState;
 
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCommitRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestUtils.java

@@ -53,7 +53,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsMana
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.ContainerAllocationExpirer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerNode;
 import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretManager;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestContinuousScheduling.java

@@ -30,7 +30,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
 import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
 import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFSAppAttempt.java

@@ -38,7 +38,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.NodeType;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
     .TestUtils;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairSchedulerPreemption.java

@@ -34,13 +34,14 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerApplicationAttempt;
 
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity
     .TestUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.ContainerPreemptEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeAddedSchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateSchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEventType;
+
+import org.apache.hadoop.yarn.server.scheduler.SchedulerRequestKey;
 import org.apache.hadoop.yarn.util.ControlledClock;
 import org.apache.hadoop.yarn.util.resource.Resources;