hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aengin...@apache.org
Subject [32/50] [abbrv] hadoop git commit: YARN-4390. Do surgical preemption based on reserved container in CapacityScheduler. Contributed by Wangda Tan
Date Mon, 09 May 2016 17:36:41 GMT
http://git-wip-us.apache.org/repos/asf/hadoop/blob/bb62e059/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
----------------------------------------------------------------------
diff --git a/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 b/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
new file mode 100644
index 0000000..1019548
--- /dev/null
+++ b/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
@@ -0,0 +1,689 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.Container;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+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.event.Dispatcher;
+import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+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.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.capacity.CSQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.ParentQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueCapacities;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;
+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.scheduler.event.ContainerPreemptEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.OrderingPolicy;
+import org.apache.hadoop.yarn.util.Clock;
+import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
+import org.apache.hadoop.yarn.util.resource.Resources;
+import org.junit.Assert;
+import org.junit.Before;
+import org.mockito.ArgumentMatcher;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeSet;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.isA;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ProportionalCapacityPreemptionPolicyMockFramework {
+  static final Log LOG =
+      LogFactory.getLog(TestProportionalCapacityPreemptionPolicyForNodePartitions.class);
+  final String ROOT = CapacitySchedulerConfiguration.ROOT;
+
+  Map<String, CSQueue> nameToCSQueues = null;
+  Map<String, Resource> partitionToResource = null;
+  Map<NodeId, FiCaSchedulerNode> nodeIdToSchedulerNodes = null;
+  RMNodeLabelsManager nlm = null;
+  RMContext rmContext = null;
+
+  ResourceCalculator rc = new DefaultResourceCalculator();
+  Clock mClock = null;
+  CapacitySchedulerConfiguration conf = null;
+  CapacityScheduler cs = null;
+  EventHandler<SchedulerEvent> mDisp = null;
+  ProportionalCapacityPreemptionPolicy policy = null;
+  Resource clusterResource = null;
+
+  @SuppressWarnings("unchecked")
+  @Before
+  public void setup() {
+    org.apache.log4j.Logger.getRootLogger().setLevel(
+        org.apache.log4j.Level.DEBUG);
+
+    conf = new CapacitySchedulerConfiguration(new Configuration(false));
+    conf.setLong(
+        CapacitySchedulerConfiguration.PREEMPTION_WAIT_TIME_BEFORE_KILL, 10000);
+    conf.setLong(CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL,
+        3000);
+    // report "ideal" preempt
+    conf.setFloat(CapacitySchedulerConfiguration.TOTAL_PREEMPTION_PER_ROUND,
+        (float) 1.0);
+    conf.setFloat(
+        CapacitySchedulerConfiguration.PREEMPTION_NATURAL_TERMINATION_FACTOR,
+        (float) 1.0);
+
+    mClock = mock(Clock.class);
+    cs = mock(CapacityScheduler.class);
+    when(cs.getResourceCalculator()).thenReturn(rc);
+    when(cs.getPreemptionManager()).thenReturn(new PreemptionManager());
+    when(cs.getConfiguration()).thenReturn(conf);
+
+    nlm = mock(RMNodeLabelsManager.class);
+    mDisp = mock(EventHandler.class);
+
+    rmContext = mock(RMContext.class);
+    when(rmContext.getNodeLabelManager()).thenReturn(nlm);
+    Dispatcher disp = mock(Dispatcher.class);
+    when(rmContext.getDispatcher()).thenReturn(disp);
+    when(disp.getEventHandler()).thenReturn(mDisp);
+    when(cs.getRMContext()).thenReturn(rmContext);
+
+    partitionToResource = new HashMap<>();
+    nodeIdToSchedulerNodes = new HashMap<>();
+    nameToCSQueues = new HashMap<>();
+  }
+
+  public void buildEnv(String labelsConfig, String nodesConfig,
+      String queuesConfig, String appsConfig) throws IOException {
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig, false);
+  }
+
+  public void buildEnv(String labelsConfig, String nodesConfig,
+      String queuesConfig, String appsConfig,
+      boolean useDominantResourceCalculator) throws IOException {
+    if (useDominantResourceCalculator) {
+      when(cs.getResourceCalculator()).thenReturn(
+          new DominantResourceCalculator());
+    }
+    mockNodeLabelsManager(labelsConfig);
+    mockSchedulerNodes(nodesConfig);
+    for (NodeId nodeId : nodeIdToSchedulerNodes.keySet()) {
+      when(cs.getSchedulerNode(nodeId)).thenReturn(
+          nodeIdToSchedulerNodes.get(nodeId));
+    }
+    List<FiCaSchedulerNode> allNodes = new ArrayList<>(
+        nodeIdToSchedulerNodes.values());
+    when(cs.getAllNodes()).thenReturn(allNodes);
+    ParentQueue root = mockQueueHierarchy(queuesConfig);
+    when(cs.getRootQueue()).thenReturn(root);
+    when(cs.getClusterResource()).thenReturn(clusterResource);
+    mockApplications(appsConfig);
+
+    policy = new ProportionalCapacityPreemptionPolicy(rmContext, cs,
+        mClock);
+  }
+
+  private void mockContainers(String containersConfig, ApplicationAttemptId attemptId,
+      String queueName, List<RMContainer> reservedContainers,
+      List<RMContainer> liveContainers) {
+    int containerId = 1;
+    int start = containersConfig.indexOf("=") + 1;
+    int end = -1;
+
+    while (start < containersConfig.length()) {
+      while (start < containersConfig.length()
+          && containersConfig.charAt(start) != '(') {
+        start++;
+      }
+      if (start >= containersConfig.length()) {
+        throw new IllegalArgumentException(
+            "Error containers specification, line=" + containersConfig);
+      }
+      end = start + 1;
+      while (end < containersConfig.length()
+          && containersConfig.charAt(end) != ')') {
+        end++;
+      }
+      if (end >= containersConfig.length()) {
+        throw new IllegalArgumentException(
+            "Error containers specification, line=" + containersConfig);
+      }
+
+      // now we found start/end, get container values
+      String[] values = containersConfig.substring(start + 1, end).split(",");
+      if (values.length != 6) {
+        throw new IllegalArgumentException("Format to define container is:"
+            + "(priority,resource,host,expression,repeat,reserved)");
+      }
+      Priority pri = Priority.newInstance(Integer.valueOf(values[0]));
+      Resource res = parseResourceFromString(values[1]);
+      NodeId host = NodeId.newInstance(values[2], 1);
+      String exp = values[3];
+      int repeat = Integer.valueOf(values[4]);
+      boolean reserved = Boolean.valueOf(values[5]);
+
+      for (int i = 0; i < repeat; i++) {
+        Container c = mock(Container.class);
+        when(c.getResource()).thenReturn(res);
+        when(c.getPriority()).thenReturn(pri);
+        RMContainerImpl rmc = mock(RMContainerImpl.class);
+        when(rmc.getAllocatedNode()).thenReturn(host);
+        when(rmc.getNodeLabelExpression()).thenReturn(exp);
+        when(rmc.getAllocatedResource()).thenReturn(res);
+        when(rmc.getContainer()).thenReturn(c);
+        when(rmc.getApplicationAttemptId()).thenReturn(attemptId);
+        when(rmc.getQueueName()).thenReturn(queueName);
+        final ContainerId cId = ContainerId.newContainerId(attemptId, containerId);
+        when(rmc.getContainerId()).thenReturn(
+            cId);
+        doAnswer(new Answer<Integer>() {
+          @Override
+          public Integer answer(InvocationOnMock invocation) throws Throwable {
+            return cId.compareTo(((RMContainer) invocation.getArguments()[0])
+                .getContainerId());
+          }
+        }).when(rmc).compareTo(any(RMContainer.class));
+
+        if (containerId == 1) {
+          when(rmc.isAMContainer()).thenReturn(true);
+        }
+
+        if (reserved) {
+          reservedContainers.add(rmc);
+          when(rmc.getReservedResource()).thenReturn(res);
+        } else {
+          liveContainers.add(rmc);
+        }
+
+        // Add container to scheduler-node
+        addContainerToSchedulerNode(host, rmc, reserved);
+
+        // If this is a non-exclusive allocation
+        String partition = null;
+        if (exp.isEmpty()
+            && !(partition = nodeIdToSchedulerNodes.get(host).getPartition())
+            .isEmpty()) {
+          LeafQueue queue = (LeafQueue) nameToCSQueues.get(queueName);
+          Map<String, TreeSet<RMContainer>> ignoreExclusivityContainers =
+              queue.getIgnoreExclusivityRMContainers();
+          if (!ignoreExclusivityContainers.containsKey(partition)) {
+            ignoreExclusivityContainers.put(partition,
+                new TreeSet<RMContainer>());
+          }
+          ignoreExclusivityContainers.get(partition).add(rmc);
+        }
+        LOG.debug("add container to app=" + attemptId + " res=" + res
+            + " node=" + host + " nodeLabelExpression=" + exp + " partition="
+            + partition);
+
+        containerId++;
+      }
+
+      start = end + 1;
+    }
+  }
+
+  /**
+   * Format is:
+   * <pre>
+   * queueName\t  // app1
+   * (priority,resource,host,expression,#repeat,reserved)
+   * (priority,resource,host,expression,#repeat,reserved);
+   * queueName\t  // app2
+   * </pre>
+   */
+  private void mockApplications(String appsConfig) {
+    int id = 1;
+    for (String a : appsConfig.split(";")) {
+      String[] strs = a.split("\t");
+      String queueName = strs[0];
+
+      // get containers
+      List<RMContainer> liveContainers = new ArrayList<RMContainer>();
+      List<RMContainer> reservedContainers = new ArrayList<RMContainer>();
+      ApplicationId appId = ApplicationId.newInstance(0L, id);
+      ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId, 1);
+
+      mockContainers(strs[1], appAttemptId, queueName, reservedContainers,
+          liveContainers);
+
+      FiCaSchedulerApp app = mock(FiCaSchedulerApp.class);
+      when(app.getLiveContainers()).thenReturn(liveContainers);
+      when(app.getReservedContainers()).thenReturn(reservedContainers);
+      when(app.getApplicationAttemptId()).thenReturn(appAttemptId);
+      when(app.getApplicationId()).thenReturn(appId);
+      when(app.getPriority()).thenReturn(Priority.newInstance(0));
+
+      // add to LeafQueue
+      LeafQueue queue = (LeafQueue) nameToCSQueues.get(queueName);
+      queue.getApplications().add(app);
+
+      id++;
+    }
+  }
+
+  private void addContainerToSchedulerNode(NodeId nodeId, RMContainer container,
+      boolean isReserved) {
+    SchedulerNode node = nodeIdToSchedulerNodes.get(nodeId);
+    assert node != null;
+
+    if (isReserved) {
+      when(node.getReservedContainer()).thenReturn(container);
+    } else {
+      node.getCopiedListOfRunningContainers().add(container);
+      Resources.subtractFrom(node.getUnallocatedResource(),
+          container.getAllocatedResource());
+    }
+  }
+
+  /**
+   * Format is:
+   * host1=partition[ res=resource];
+   * host2=partition[ res=resource];
+   */
+  private void mockSchedulerNodes(String schedulerNodesConfigStr)
+      throws IOException {
+    String[] nodesConfigStrArray = schedulerNodesConfigStr.split(";");
+    for (String p : nodesConfigStrArray) {
+      String[] arr = p.split(" ");
+
+      NodeId nodeId = NodeId.newInstance(arr[0].substring(0, arr[0].indexOf("=")), 1);
+      String partition = arr[0].substring(arr[0].indexOf("=") + 1, arr[0].length());
+
+      FiCaSchedulerNode sn = mock(FiCaSchedulerNode.class);
+      when(sn.getNodeID()).thenReturn(nodeId);
+      when(sn.getPartition()).thenReturn(partition);
+
+      Resource totalRes = Resources.createResource(0);
+      if (arr.length > 1) {
+        String res = arr[1];
+        if (res.contains("res=")) {
+          String resSring = res.substring(
+              res.indexOf("res=") + "res=".length());
+          totalRes = parseResourceFromString(resSring);
+        }
+      }
+      when(sn.getTotalResource()).thenReturn(totalRes);
+      when(sn.getUnallocatedResource()).thenReturn(Resources.clone(totalRes));
+
+      // TODO, add settings of killable resources when necessary
+      when(sn.getTotalKillableResources()).thenReturn(Resources.none());
+
+      List<RMContainer> liveContainers = new ArrayList<>();
+      when(sn.getCopiedListOfRunningContainers()).thenReturn(liveContainers);
+
+      nodeIdToSchedulerNodes.put(nodeId, sn);
+
+      LOG.debug("add scheduler node, id=" + nodeId + ", partition=" + partition);
+    }
+  }
+
+  /**
+   * Format is:
+   * <pre>
+   * partition0=total_resource,exclusivity;
+   * partition1=total_resource,exclusivity;
+   * ...
+   * </pre>
+   */
+  private void mockNodeLabelsManager(String nodeLabelsConfigStr) throws IOException {
+    String[] partitionConfigArr = nodeLabelsConfigStr.split(";");
+    clusterResource = Resources.createResource(0);
+    for (String p : partitionConfigArr) {
+      String partitionName = p.substring(0, p.indexOf("="));
+      Resource res = parseResourceFromString(p.substring(p.indexOf("=") + 1,
+          p.indexOf(",")));
+      boolean exclusivity =
+          Boolean.valueOf(p.substring(p.indexOf(",") + 1, p.length()));
+      when(nlm.getResourceByLabel(eq(partitionName), any(Resource.class)))
+          .thenReturn(res);
+      when(nlm.isExclusiveNodeLabel(eq(partitionName))).thenReturn(exclusivity);
+
+      // add to partition to resource
+      partitionToResource.put(partitionName, res);
+      LOG.debug("add partition=" + partitionName + " totalRes=" + res
+          + " exclusivity=" + exclusivity);
+      Resources.addTo(clusterResource, res);
+    }
+
+    when(nlm.getClusterNodeLabelNames()).thenReturn(
+        partitionToResource.keySet());
+  }
+
+  private Resource parseResourceFromString(String p) {
+    String[] resource = p.split(":");
+    Resource res;
+    if (resource.length == 1) {
+      res = Resources.createResource(Integer.valueOf(resource[0]));
+    } else {
+      res = Resources.createResource(Integer.valueOf(resource[0]),
+          Integer.valueOf(resource[1]));
+    }
+    return res;
+  }
+
+  /**
+   * Format is:
+   * <pre>
+   * root (<partition-name-1>=[guaranteed max used pending (reserved)],<partition-name-2>=..);
+   * -A(...);
+   * --A1(...);
+   * --A2(...);
+   * -B...
+   * </pre>
+   * ";" splits queues, and there should no empty lines, no extra spaces
+   */
+  @SuppressWarnings({ "unchecked", "rawtypes" })
+  private ParentQueue mockQueueHierarchy(String queueExprs) {
+    String[] queueExprArray = queueExprs.split(";");
+    ParentQueue rootQueue = null;
+    for (int idx = 0; idx < queueExprArray.length; idx++) {
+      String q = queueExprArray[idx];
+      CSQueue queue;
+
+      // Initialize queue
+      if (isParent(queueExprArray, idx)) {
+        ParentQueue parentQueue = mock(ParentQueue.class);
+        queue = parentQueue;
+        List<CSQueue> children = new ArrayList<CSQueue>();
+        when(parentQueue.getChildQueues()).thenReturn(children);
+      } else {
+        LeafQueue leafQueue = mock(LeafQueue.class);
+        final TreeSet<FiCaSchedulerApp> apps = new TreeSet<>(
+            new Comparator<FiCaSchedulerApp>() {
+              @Override
+              public int compare(FiCaSchedulerApp a1, FiCaSchedulerApp a2) {
+                return a1.getApplicationId().compareTo(a2.getApplicationId());
+              }
+            });
+        when(leafQueue.getApplications()).thenReturn(apps);
+        OrderingPolicy<FiCaSchedulerApp> so = mock(OrderingPolicy.class);
+        when(so.getPreemptionIterator()).thenAnswer(new Answer() {
+          public Object answer(InvocationOnMock invocation) {
+            return apps.descendingIterator();
+          }
+        });
+        when(leafQueue.getOrderingPolicy()).thenReturn(so);
+
+        Map<String, TreeSet<RMContainer>> ignorePartitionContainers =
+            new HashMap<>();
+        when(leafQueue.getIgnoreExclusivityRMContainers()).thenReturn(
+            ignorePartitionContainers);
+        queue = leafQueue;
+      }
+
+      setupQueue(queue, q, queueExprArray, idx);
+      if (queue.getQueueName().equals(ROOT)) {
+        rootQueue = (ParentQueue) queue;
+      }
+    }
+    return rootQueue;
+  }
+
+  private void setupQueue(CSQueue queue, String q, String[] queueExprArray,
+      int idx) {
+    LOG.debug("*** Setup queue, source=" + q);
+    String queuePath = null;
+
+    int myLevel = getLevel(q);
+    if (0 == myLevel) {
+      // It's root
+      when(queue.getQueueName()).thenReturn(ROOT);
+      queuePath = ROOT;
+    }
+
+    String queueName = getQueueName(q);
+    when(queue.getQueueName()).thenReturn(queueName);
+
+    // Setup parent queue, and add myself to parentQueue.children-list
+    ParentQueue parentQueue = getParentQueue(queueExprArray, idx, myLevel);
+    if (null != parentQueue) {
+      when(queue.getParent()).thenReturn(parentQueue);
+      parentQueue.getChildQueues().add(queue);
+
+      // Setup my path
+      queuePath = parentQueue.getQueuePath() + "." + queueName;
+    }
+    when(queue.getQueuePath()).thenReturn(queuePath);
+
+    QueueCapacities qc = new QueueCapacities(0 == myLevel);
+    ResourceUsage ru = new ResourceUsage();
+
+    when(queue.getQueueCapacities()).thenReturn(qc);
+    when(queue.getQueueResourceUsage()).thenReturn(ru);
+
+    LOG.debug("Setup queue, name=" + queue.getQueueName() + " path="
+        + queue.getQueuePath());
+    LOG.debug("Parent=" + (parentQueue == null ? "null" : parentQueue
+        .getQueueName()));
+
+    // Setup other fields like used resource, guaranteed resource, etc.
+    String capacitySettingStr = q.substring(q.indexOf("(") + 1, q.indexOf(")"));
+    for (String s : capacitySettingStr.split(",")) {
+      String partitionName = s.substring(0, s.indexOf("="));
+      String[] values = s.substring(s.indexOf("[") + 1, s.indexOf("]")).split(" ");
+      // Add a small epsilon to capacities to avoid truncate when doing
+      // Resources.multiply
+      float epsilon = 1e-6f;
+      Resource totResoucePerPartition = partitionToResource.get(partitionName);
+      float absGuaranteed = Resources.divide(rc, totResoucePerPartition,
+          parseResourceFromString(values[0].trim()), totResoucePerPartition)
+          + epsilon;
+      float absMax = Resources.divide(rc, totResoucePerPartition,
+          parseResourceFromString(values[1].trim()), totResoucePerPartition)
+          + epsilon;
+      float absUsed = Resources.divide(rc, totResoucePerPartition,
+          parseResourceFromString(values[2].trim()), totResoucePerPartition)
+          + epsilon;
+      Resource pending = parseResourceFromString(values[3].trim());
+      qc.setAbsoluteCapacity(partitionName, absGuaranteed);
+      qc.setAbsoluteMaximumCapacity(partitionName, absMax);
+      qc.setAbsoluteUsedCapacity(partitionName, absUsed);
+      ru.setPending(partitionName, pending);
+      if (!isParent(queueExprArray, idx)) {
+        LeafQueue lq = (LeafQueue) queue;
+        when(lq.getTotalPendingResourcesConsideringUserLimit(isA(Resource.class),
+            isA(String.class))).thenReturn(pending);
+      }
+      ru.setUsed(partitionName, parseResourceFromString(values[2].trim()));
+
+      // Setup reserved resource if it contained by input config
+      Resource reserved = Resources.none();
+      if(values.length == 5) {
+        reserved = parseResourceFromString(values[4].trim());
+        ru.setReserved(partitionName, reserved);
+      }
+      LOG.debug("Setup queue=" + queueName + " partition=" + partitionName
+          + " [abs_guaranteed=" + absGuaranteed + ",abs_max=" + absMax
+          + ",abs_used" + absUsed + ",pending_resource=" + pending
+          + ", reserved_resource=" + reserved + "]");
+    }
+
+    // Setup preemption disabled
+    when(queue.getPreemptionDisabled()).thenReturn(
+        conf.getPreemptionDisabled(queuePath, false));
+
+    nameToCSQueues.put(queueName, queue);
+    when(cs.getQueue(eq(queueName))).thenReturn(queue);
+  }
+
+  /**
+   * Level of a queue is how many "-" at beginning, root's level is 0
+   */
+  private int getLevel(String q) {
+    int level = 0; // level = how many "-" at beginning
+    while (level < q.length() && q.charAt(level) == '-') {
+      level++;
+    }
+    return level;
+  }
+
+  private String getQueueName(String q) {
+    int idx = 0;
+    // find first != '-' char
+    while (idx < q.length() && q.charAt(idx) == '-') {
+      idx++;
+    }
+    if (idx == q.length()) {
+      throw new IllegalArgumentException("illegal input:" + q);
+    }
+    // name = after '-' and before '('
+    String name = q.substring(idx, q.indexOf('('));
+    if (name.isEmpty()) {
+      throw new IllegalArgumentException("queue name shouldn't be empty:" + q);
+    }
+    if (name.contains(".")) {
+      throw new IllegalArgumentException("queue name shouldn't contain '.':"
+          + name);
+    }
+    return name;
+  }
+
+  private ParentQueue getParentQueue(String[] queueExprArray, int idx, int myLevel) {
+    idx--;
+    while (idx >= 0) {
+      int level = getLevel(queueExprArray[idx]);
+      if (level < myLevel) {
+        String parentQueuName = getQueueName(queueExprArray[idx]);
+        return (ParentQueue) nameToCSQueues.get(parentQueuName);
+      }
+      idx--;
+    }
+
+    return null;
+  }
+
+  /**
+   * Get if a queue is ParentQueue
+   */
+  private boolean isParent(String[] queues, int idx) {
+    int myLevel = getLevel(queues[idx]);
+    idx++;
+    while (idx < queues.length && getLevel(queues[idx]) == myLevel) {
+      idx++;
+    }
+    if (idx >= queues.length || getLevel(queues[idx]) < myLevel) {
+      // It's a LeafQueue
+      return false;
+    } else {
+      return true;
+    }
+  }
+
+  public ApplicationAttemptId getAppAttemptId(int id) {
+    ApplicationId appId = ApplicationId.newInstance(0L, id);
+    ApplicationAttemptId appAttemptId =
+        ApplicationAttemptId.newInstance(appId, 1);
+    return appAttemptId;
+  }
+
+  public void checkContainerNodesInApp(FiCaSchedulerApp app,
+      int expectedContainersNumber, String host) {
+    NodeId nodeId = NodeId.newInstance(host, 1);
+    int num = 0;
+    for (RMContainer c : app.getLiveContainers()) {
+      if (c.getAllocatedNode().equals(nodeId)) {
+        num++;
+      }
+    }
+    for (RMContainer c : app.getReservedContainers()) {
+      if (c.getAllocatedNode().equals(nodeId)) {
+        num++;
+      }
+    }
+    Assert.assertEquals(expectedContainersNumber, num);
+  }
+
+  public FiCaSchedulerApp getApp(String queueName, int appId) {
+    for (FiCaSchedulerApp app : ((LeafQueue) cs.getQueue(queueName))
+        .getApplications()) {
+      if (app.getApplicationId().getId() == appId) {
+        return app;
+      }
+    }
+    return null;
+  }
+
+  public void checkAbsCapacities(CSQueue queue, String partition,
+      float guaranteed, float max, float used) {
+    QueueCapacities qc = queue.getQueueCapacities();
+    Assert.assertEquals(guaranteed, qc.getAbsoluteCapacity(partition), 1e-3);
+    Assert.assertEquals(max, qc.getAbsoluteMaximumCapacity(partition), 1e-3);
+    Assert.assertEquals(used, qc.getAbsoluteUsedCapacity(partition), 1e-3);
+  }
+
+  public void checkPendingResource(CSQueue queue, String partition, int pending) {
+    ResourceUsage ru = queue.getQueueResourceUsage();
+    Assert.assertEquals(pending, ru.getPending(partition).getMemory());
+  }
+
+  public void checkReservedResource(CSQueue queue, String partition, int reserved) {
+    ResourceUsage ru = queue.getQueueResourceUsage();
+    Assert.assertEquals(reserved, ru.getReserved(partition).getMemory());
+  }
+
+  static class IsPreemptionRequestForQueueAndNode
+      extends ArgumentMatcher<ContainerPreemptEvent> {
+    private final ApplicationAttemptId appAttId;
+    private final String queueName;
+    private final NodeId nodeId;
+
+    IsPreemptionRequestForQueueAndNode(ApplicationAttemptId appAttId,
+        String queueName, NodeId nodeId) {
+      this.appAttId = appAttId;
+      this.queueName = queueName;
+      this.nodeId = nodeId;
+    }
+    @Override
+    public boolean matches(Object o) {
+      ContainerPreemptEvent cpe = (ContainerPreemptEvent)o;
+
+      return appAttId.equals(cpe.getAppId())
+          && queueName.equals(cpe.getContainer().getQueueName())
+          && nodeId.equals(cpe.getContainer().getAllocatedNode());
+    }
+    @Override
+    public String toString() {
+      return appAttId.toString();
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/bb62e059/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
----------------------------------------------------------------------
diff --git a/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 b/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
index 499a3d0..5b7ac52 100644
--- a/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
+++ b/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
@@ -62,6 +62,7 @@ import org.mockito.ArgumentMatcher;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.Deque;
@@ -161,6 +162,11 @@ public class TestProportionalCapacityPreemptionPolicy {
     mCS = mock(CapacityScheduler.class);
     when(mCS.getResourceCalculator()).thenReturn(rc);
     lm = mock(RMNodeLabelsManager.class);
+    try {
+      when(lm.isExclusiveNodeLabel(anyString())).thenReturn(true);
+    } catch (IOException e) {
+      // do nothing
+    }
     when(mCS.getConfiguration()).thenReturn(conf);
     rmContext = mock(RMContext.class);
     when(mCS.getRMContext()).thenReturn(rmContext);
@@ -650,6 +656,26 @@ public class TestProportionalCapacityPreemptionPolicy {
   }
 
   @Test
+  public void testHierarchicalWithReserved() {
+    int[][] qData = new int[][] {
+        //  /    A   B   C    D   E   F
+        { 200, 100, 50, 50, 100, 10, 90 },  // abs
+        { 200, 200, 200, 200, 200, 200, 200 },  // maxCap
+        { 200, 110, 60, 50,  90, 90,  0 },  // used
+        {  10,   0,  0,  0,  10,  0, 10 },  // pending
+        {  40,  25,  15, 10, 15,  15,  0 },  // reserved
+        {   4,   2,  1,  1,   2,  1,  1 },  // apps
+        {  -1,  -1,  1,  1,  -1,  1,  1 },  // req granularity
+        {   2,   2,  0,  0,   2,  0,  0 },  // subqueues
+    };
+    ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData);
+    policy.editSchedule();
+    // verify capacity taken from A1, not B1 despite B1 being far over
+    // its absolute guaranteed capacity
+    verify(mDisp, times(10)).handle(argThat(new IsPreemptionRequestFor(appA)));
+  }
+
+  @Test
   public void testZeroGuar() {
     int[][] qData = new int[][] {
       //  /    A   B   C    D   E   F
@@ -934,7 +960,38 @@ public class TestProportionalCapacityPreemptionPolicy {
     //check the parent can prempt only the extra from > 2 level child
     TempQueuePerPartition tempQueueAPartition = policy.getQueuePartitions().get("queueA").get("");
     assertEquals(0, tempQueueAPartition.untouchableExtra.getMemory());
-    int extraForQueueA = tempQueueAPartition.current.getMemory()- tempQueueAPartition.guaranteed.getMemory();
+    int extraForQueueA = tempQueueAPartition.getUsed().getMemory()
+        - tempQueueAPartition.getGuaranteed().getMemory();
+    assertEquals(extraForQueueA,tempQueueAPartition.preemptableExtra.getMemory());
+  }
+
+  @Test
+  public void testHierarchicalLarge3LevelsWithReserved() {
+    int[][] qData = new int[][] {
+        //  /    A                      F               I
+        //            B    C                  G    H          J    K
+        //                    D    E
+        { 400, 200,  60, 140, 100, 40,  100,  70,  30, 100,  10,  90 },  // abs
+        { 400, 400, 400, 400, 400, 400, 400, 400, 400, 400, 400, 400 },  // maxCap
+        { 400, 210,  60, 150, 100, 50,  100,  50,  50,  90,  10,   80 },  // used
+        {  10,   0,   0,   0,  0,   0,   0,   0,   0,   0,   0,  10 },  // pending
+        {  50,  30,  20,   10, 5,   5,   0,   0,   0,  10,  10,   0 },  // reserved
+        //          appA     appB appC   appD appE      appF appG
+        {   7,   3,   1,   2,   1,   1,  2,   1,   1,   2,   1,   1 },  // apps
+        {  -1,  -1,   1,   -1,  1,   1,  -1,   1,   1,  -1,   1,   1 },  // req granularity
+        {   3,   2,   0,   2,   0,   0,   2,   0,   0,   2,   0,   0 },  // subqueues
+    };
+    ProportionalCapacityPreemptionPolicy policy = buildPolicy(qData);
+    policy.editSchedule();
+
+    verify(mDisp, times(10)).handle(argThat(new IsPreemptionRequestFor(appC)));
+    assertEquals(10, policy.getQueuePartitions().get("queueE").get("").preemptableExtra.getMemory());
+    //2nd level child(E) preempts 10, but parent A has only 9 extra
+    //check the parent can prempt only the extra from > 2 level child
+    TempQueuePerPartition tempQueueAPartition = policy.getQueuePartitions().get("queueA").get("");
+    assertEquals(0, tempQueueAPartition.untouchableExtra.getMemory());
+    int extraForQueueA = tempQueueAPartition.getUsed().getMemory()
+        - tempQueueAPartition.getGuaranteed().getMemory();
     assertEquals(extraForQueueA,tempQueueAPartition.preemptableExtra.getMemory());
   }
 
@@ -1058,6 +1115,7 @@ public class TestProportionalCapacityPreemptionPolicy {
     ParentQueue root = mockParentQueue(null, queues[0], pqs);
     ResourceUsage resUsage = new ResourceUsage();
     resUsage.setUsed(used[0]);
+    resUsage.setReserved(reserved[0]);
     when(root.getQueueName()).thenReturn(CapacitySchedulerConfiguration.ROOT);
     when(root.getAbsoluteUsedCapacity()).thenReturn(
         Resources.divide(rc, tot, used[0], tot));
@@ -1083,6 +1141,7 @@ public class TestProportionalCapacityPreemptionPolicy {
         q = mockParentQueue(p, queues[i], pqs);
         ResourceUsage resUsagePerQueue = new ResourceUsage();
         resUsagePerQueue.setUsed(used[i]);
+        resUsagePerQueue.setReserved(reserved[i]);
         when(q.getQueueResourceUsage()).thenReturn(resUsagePerQueue);
       } else {
         q = mockLeafQueue(p, tot, i, abs, used, pending, reserved, apps, gran);
@@ -1159,6 +1218,7 @@ public class TestProportionalCapacityPreemptionPolicy {
     ResourceUsage ru = new ResourceUsage();
     ru.setPending(pending[i]);
     ru.setUsed(used[i]);
+    ru.setReserved(reserved[i]);
     when(lq.getQueueResourceUsage()).thenReturn(ru);
     // consider moving where CapacityScheduler::comparator accessible
     final NavigableSet<FiCaSchedulerApp> qApps = new TreeSet<FiCaSchedulerApp>(

http://git-wip-us.apache.org/repos/asf/hadoop/blob/bb62e059/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForNodePartitions.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForNodePartitions.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForNodePartitions.java
index 5ffae6e..e31a889 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForNodePartitions.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForNodePartitions.java
@@ -18,231 +18,23 @@
 
 package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
-import org.apache.hadoop.yarn.api.records.ApplicationId;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-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.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.event.Dispatcher;
-import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity.TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor;
-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.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.capacity.CSQueue;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.LeafQueue;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.ParentQueue;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.QueueCapacities;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.preemption.PreemptionManager;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.fica.FiCaSchedulerApp;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
-import org.apache.hadoop.yarn.server.resourcemanager.scheduler.policy.OrderingPolicy;
-import org.apache.hadoop.yarn.util.Clock;
-import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
-import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
-import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
-import org.apache.hadoop.yarn.util.resource.Resources;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
 
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Comparator;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.TreeSet;
 
-import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.argThat;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Matchers.isA;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-public class TestProportionalCapacityPreemptionPolicyForNodePartitions {
-  private static final Log LOG =
-      LogFactory.getLog(TestProportionalCapacityPreemptionPolicyForNodePartitions.class);
-  static final String ROOT = CapacitySchedulerConfiguration.ROOT;
 
-  private Map<String, CSQueue> nameToCSQueues = null;
-  private Map<String, Resource> partitionToResource = null;
-  private Map<NodeId, SchedulerNode> nodeIdToSchedulerNodes = null;
-  private RMNodeLabelsManager nlm = null;
-  private RMContext rmContext = null;
-
-  private ResourceCalculator rc = new DefaultResourceCalculator();
-  private Clock mClock = null;
-  private CapacitySchedulerConfiguration conf = null;
-  private CapacityScheduler cs = null;
-  private EventHandler<SchedulerEvent> mDisp = null;
-  private ProportionalCapacityPreemptionPolicy policy = null;
-  private Resource clusterResource = null;
-
-  @SuppressWarnings("unchecked")
+public class TestProportionalCapacityPreemptionPolicyForNodePartitions
+    extends ProportionalCapacityPreemptionPolicyMockFramework {
   @Before
   public void setup() {
-    org.apache.log4j.Logger.getRootLogger().setLevel(
-        org.apache.log4j.Level.DEBUG);
-
-    conf = new CapacitySchedulerConfiguration(new Configuration(false));
-    conf.setLong(
-        CapacitySchedulerConfiguration.PREEMPTION_WAIT_TIME_BEFORE_KILL, 10000);
-    conf.setLong(CapacitySchedulerConfiguration.PREEMPTION_MONITORING_INTERVAL,
-        3000);
-    // report "ideal" preempt
-    conf.setFloat(CapacitySchedulerConfiguration.TOTAL_PREEMPTION_PER_ROUND,
-        (float) 1.0);
-    conf.setFloat(
-        CapacitySchedulerConfiguration.PREEMPTION_NATURAL_TERMINATION_FACTOR,
-        (float) 1.0);
-
-    mClock = mock(Clock.class);
-    cs = mock(CapacityScheduler.class);
-    when(cs.getResourceCalculator()).thenReturn(rc);
-    when(cs.getPreemptionManager()).thenReturn(new PreemptionManager());
-    when(cs.getConfiguration()).thenReturn(conf);
-
-    nlm = mock(RMNodeLabelsManager.class);
-    mDisp = mock(EventHandler.class);
-
-    rmContext = mock(RMContext.class);
-    when(rmContext.getNodeLabelManager()).thenReturn(nlm);
-    Dispatcher disp = mock(Dispatcher.class);
-    when(rmContext.getDispatcher()).thenReturn(disp);
-    when(disp.getEventHandler()).thenReturn(mDisp);
-    when(cs.getRMContext()).thenReturn(rmContext);
-
+    super.setup();
     policy = new ProportionalCapacityPreemptionPolicy(rmContext, cs, mClock);
-    partitionToResource = new HashMap<>();
-    nodeIdToSchedulerNodes = new HashMap<>();
-    nameToCSQueues = new HashMap<>();
-  }
-
-  @Test
-  public void testBuilder() throws Exception {
-    /**
-     * Test of test, make sure we build expected mock schedulable objects
-     */
-    String labelsConfig =
-        "=200,true;" + // default partition
-        "red=100,false;" + // partition=red
-        "blue=200,true"; // partition=blue
-    String nodesConfig =
-        "n1=red;" + // n1 has partition=red
-        "n2=blue;" + // n2 has partition=blue
-        "n3="; // n3 doesn't have partition
-    String queuesConfig =
-        // guaranteed,max,used,pending
-        "root(=[200 200 100 100],red=[100 100 100 100],blue=[200 200 200 200]);" + //root
-        "-a(=[100 200 100 100],red=[0 0 0 0],blue=[200 200 200 200]);" + // a
-        "--a1(=[50 100 50 100],red=[0 0 0 0],blue=[100 200 200 0]);" + // a1
-        "--a2(=[50 200 50 0],red=[0 0 0 0],blue=[100 200 0 200]);" + // a2
-        "-b(=[100 200 0 0],red=[100 100 100 100],blue=[0 0 0 0])";
-    String appsConfig=
-        //queueName\t(priority,resource,host,expression,#repeat,reserved)
-        // app1 in a1, , 50 in n2 (reserved), 50 in n2 (allocated)
-        "a1\t" // app1 in a1
-        + "(1,1,n3,red,50,false);" + // 50 * default in n3
-
-        "a1\t" // app2 in a1
-        + "(2,1,n2,,50,true)(2,1,n2,,50,false)" // 50 * ignore-exclusivity (reserved),
-                                                // 50 * ignore-exclusivity (allocated)
-        + "(2,1,n2,blue,50,true)(2,1,n2,blue,50,true);" + // 50 in n2 (reserved),
-                                                          // 50 in n2 (allocated)
-        "a2\t" // app3 in a2
-        + "(1,1,n3,red,50,false);" + // 50 * default in n3
-
-        "b\t" // app4 in b
-        + "(1,1,n1,red,100,false);";
-
-    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
-
-    // Check queues:
-    // root
-    checkAbsCapacities(cs.getQueue("root"), "", 1f, 1f, 0.5f);
-    checkPendingResource(cs.getQueue("root"), "", 100);
-    checkAbsCapacities(cs.getQueue("root"), "red", 1f, 1f, 1f);
-    checkPendingResource(cs.getQueue("root"), "red", 100);
-    checkAbsCapacities(cs.getQueue("root"), "blue", 1f, 1f, 1f);
-    checkPendingResource(cs.getQueue("root"), "blue", 200);
-
-    // a
-    checkAbsCapacities(cs.getQueue("a"), "", 0.5f, 1f, 0.5f);
-    checkPendingResource(cs.getQueue("a"), "", 100);
-    checkAbsCapacities(cs.getQueue("a"), "red", 0f, 0f, 0f);
-    checkPendingResource(cs.getQueue("a"), "red", 0);
-    checkAbsCapacities(cs.getQueue("a"), "blue", 1f, 1f, 1f);
-    checkPendingResource(cs.getQueue("a"), "blue", 200);
-
-    // a1
-    checkAbsCapacities(cs.getQueue("a1"), "", 0.25f, 0.5f, 0.25f);
-    checkPendingResource(cs.getQueue("a1"), "", 100);
-    checkAbsCapacities(cs.getQueue("a1"), "red", 0f, 0f, 0f);
-    checkPendingResource(cs.getQueue("a1"), "red", 0);
-    checkAbsCapacities(cs.getQueue("a1"), "blue", 0.5f, 1f, 1f);
-    checkPendingResource(cs.getQueue("a1"), "blue", 0);
-
-    // a2
-    checkAbsCapacities(cs.getQueue("a2"), "", 0.25f, 1f, 0.25f);
-    checkPendingResource(cs.getQueue("a2"), "", 0);
-    checkAbsCapacities(cs.getQueue("a2"), "red", 0f, 0f, 0f);
-    checkPendingResource(cs.getQueue("a2"), "red", 0);
-    checkAbsCapacities(cs.getQueue("a2"), "blue", 0.5f, 1f, 0f);
-    checkPendingResource(cs.getQueue("a2"), "blue", 200);
-
-    // b1
-    checkAbsCapacities(cs.getQueue("b"), "", 0.5f, 1f, 0f);
-    checkPendingResource(cs.getQueue("b"), "", 0);
-    checkAbsCapacities(cs.getQueue("b"), "red", 1f, 1f, 1f);
-    checkPendingResource(cs.getQueue("b"), "red", 100);
-    checkAbsCapacities(cs.getQueue("b"), "blue", 0f, 0f, 0f);
-    checkPendingResource(cs.getQueue("b"), "blue", 0);
-
-    // Check ignored partitioned containers in queue
-    Assert.assertEquals(100, ((LeafQueue) cs.getQueue("a1"))
-        .getIgnoreExclusivityRMContainers().get("blue").size());
-
-    // Check applications
-    Assert.assertEquals(2, ((LeafQueue)cs.getQueue("a1")).getApplications().size());
-    Assert.assertEquals(1, ((LeafQueue)cs.getQueue("a2")).getApplications().size());
-    Assert.assertEquals(1, ((LeafQueue)cs.getQueue("b")).getApplications().size());
-
-    // Check #containers
-    FiCaSchedulerApp app1 = getApp("a1", 1);
-    FiCaSchedulerApp app2 = getApp("a1", 2);
-    FiCaSchedulerApp app3 = getApp("a2", 3);
-    FiCaSchedulerApp app4 = getApp("b", 4);
-
-    Assert.assertEquals(50, app1.getLiveContainers().size());
-    checkContainerNodesInApp(app1, 50, "n3");
-
-    Assert.assertEquals(50, app2.getLiveContainers().size());
-    Assert.assertEquals(150, app2.getReservedContainers().size());
-    checkContainerNodesInApp(app2, 200, "n2");
-
-    Assert.assertEquals(50, app3.getLiveContainers().size());
-    checkContainerNodesInApp(app3, 50, "n3");
-
-    Assert.assertEquals(100, app4.getLiveContainers().size());
-    checkContainerNodesInApp(app4, 100, "n1");
   }
 
   @Test
@@ -822,477 +614,4 @@ public class TestProportionalCapacityPreemptionPolicyForNodePartitions {
     verify(mDisp, never()).handle(
         argThat(new IsPreemptionRequestFor(getAppAttemptId(3))));
   }
-
-  private ApplicationAttemptId getAppAttemptId(int id) {
-    ApplicationId appId = ApplicationId.newInstance(0L, id);
-    ApplicationAttemptId appAttemptId =
-        ApplicationAttemptId.newInstance(appId, 1);
-    return appAttemptId;
-  }
-
-  private void checkContainerNodesInApp(FiCaSchedulerApp app,
-      int expectedContainersNumber, String host) {
-    NodeId nodeId = NodeId.newInstance(host, 1);
-    int num = 0;
-    for (RMContainer c : app.getLiveContainers()) {
-      if (c.getAllocatedNode().equals(nodeId)) {
-        num++;
-      }
-    }
-    for (RMContainer c : app.getReservedContainers()) {
-      if (c.getAllocatedNode().equals(nodeId)) {
-        num++;
-      }
-    }
-    Assert.assertEquals(expectedContainersNumber, num);
-  }
-
-  private FiCaSchedulerApp getApp(String queueName, int appId) {
-    for (FiCaSchedulerApp app : ((LeafQueue) cs.getQueue(queueName))
-        .getApplications()) {
-      if (app.getApplicationId().getId() == appId) {
-        return app;
-      }
-    }
-    return null;
-  }
-
-  private void checkAbsCapacities(CSQueue queue, String partition,
-      float guaranteed, float max, float used) {
-    QueueCapacities qc = queue.getQueueCapacities();
-    Assert.assertEquals(guaranteed, qc.getAbsoluteCapacity(partition), 1e-3);
-    Assert.assertEquals(max, qc.getAbsoluteMaximumCapacity(partition), 1e-3);
-    Assert.assertEquals(used, qc.getAbsoluteUsedCapacity(partition), 1e-3);
-  }
-
-  private void checkPendingResource(CSQueue queue, String partition, int pending) {
-    ResourceUsage ru = queue.getQueueResourceUsage();
-    Assert.assertEquals(pending, ru.getPending(partition).getMemory());
-  }
-
-  private void buildEnv(String labelsConfig, String nodesConfig,
-      String queuesConfig, String appsConfig) throws IOException {
-    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig, false);
-  }
-
-  private void buildEnv(String labelsConfig, String nodesConfig,
-      String queuesConfig, String appsConfig,
-      boolean useDominantResourceCalculator) throws IOException {
-    if (useDominantResourceCalculator) {
-      when(cs.getResourceCalculator()).thenReturn(
-          new DominantResourceCalculator());
-    }
-    mockNodeLabelsManager(labelsConfig);
-    mockSchedulerNodes(nodesConfig);
-    for (NodeId nodeId : nodeIdToSchedulerNodes.keySet()) {
-      when(cs.getSchedulerNode(nodeId)).thenReturn(
-          nodeIdToSchedulerNodes.get(nodeId));
-    }
-    ParentQueue root = mockQueueHierarchy(queuesConfig);
-    when(cs.getRootQueue()).thenReturn(root);
-    when(cs.getClusterResource()).thenReturn(clusterResource);
-    mockApplications(appsConfig);
-
-    policy = new ProportionalCapacityPreemptionPolicy(rmContext, cs,
-        mClock);
-  }
-
-  private void mockContainers(String containersConfig, ApplicationAttemptId attemptId,
-      String queueName, List<RMContainer> reservedContainers,
-      List<RMContainer> liveContainers) {
-    int containerId = 1;
-    int start = containersConfig.indexOf("=") + 1;
-    int end = -1;
-
-    while (start < containersConfig.length()) {
-      while (start < containersConfig.length()
-          && containersConfig.charAt(start) != '(') {
-        start++;
-      }
-      if (start >= containersConfig.length()) {
-        throw new IllegalArgumentException(
-            "Error containers specification, line=" + containersConfig);
-      }
-      end = start + 1;
-      while (end < containersConfig.length()
-          && containersConfig.charAt(end) != ')') {
-        end++;
-      }
-      if (end >= containersConfig.length()) {
-        throw new IllegalArgumentException(
-            "Error containers specification, line=" + containersConfig);
-      }
-
-      // now we found start/end, get container values
-      String[] values = containersConfig.substring(start + 1, end).split(",");
-      if (values.length != 6) {
-        throw new IllegalArgumentException("Format to define container is:"
-            + "(priority,resource,host,expression,repeat,reserved)");
-      }
-      Priority pri = Priority.newInstance(Integer.parseInt(values[0]));
-      Resource res = parseResourceFromString(values[1]);
-      NodeId host = NodeId.newInstance(values[2], 1);
-      String exp = values[3];
-      int repeat = Integer.parseInt(values[4]);
-      boolean reserved = Boolean.parseBoolean(values[5]);
-
-      for (int i = 0; i < repeat; i++) {
-        Container c = mock(Container.class);
-        when(c.getResource()).thenReturn(res);
-        when(c.getPriority()).thenReturn(pri);
-        RMContainerImpl rmc = mock(RMContainerImpl.class);
-        when(rmc.getAllocatedNode()).thenReturn(host);
-        when(rmc.getNodeLabelExpression()).thenReturn(exp);
-        when(rmc.getAllocatedResource()).thenReturn(res);
-        when(rmc.getContainer()).thenReturn(c);
-        when(rmc.getApplicationAttemptId()).thenReturn(attemptId);
-        final ContainerId cId = ContainerId.newContainerId(attemptId, containerId);
-        when(rmc.getContainerId()).thenReturn(
-            cId);
-        doAnswer(new Answer<Integer>() {
-          @Override
-          public Integer answer(InvocationOnMock invocation) throws Throwable {
-            return cId.compareTo(((RMContainer) invocation.getArguments()[0])
-                .getContainerId());
-          }
-        }).when(rmc).compareTo(any(RMContainer.class));
-
-        if (containerId == 1) {
-          when(rmc.isAMContainer()).thenReturn(true);
-        }
-
-        if (reserved) {
-          reservedContainers.add(rmc);
-        } else {
-          liveContainers.add(rmc);
-        }
-
-        // If this is a non-exclusive allocation
-        String partition = null;
-        if (exp.isEmpty()
-            && !(partition = nodeIdToSchedulerNodes.get(host).getPartition())
-                .isEmpty()) {
-          LeafQueue queue = (LeafQueue) nameToCSQueues.get(queueName);
-          Map<String, TreeSet<RMContainer>> ignoreExclusivityContainers =
-              queue.getIgnoreExclusivityRMContainers();
-          if (!ignoreExclusivityContainers.containsKey(partition)) {
-            ignoreExclusivityContainers.put(partition,
-                new TreeSet<RMContainer>());
-          }
-          ignoreExclusivityContainers.get(partition).add(rmc);
-        }
-        LOG.debug("add container to app=" + attemptId + " res=" + res
-            + " node=" + host + " nodeLabelExpression=" + exp + " partition="
-            + partition);
-
-        containerId++;
-      }
-
-      start = end + 1;
-    }
-  }
-
-  /**
-   * Format is:
-   * <pre>
-   * queueName\t  // app1
-   * (priority,resource,host,expression,#repeat,reserved)
-   * (priority,resource,host,expression,#repeat,reserved);
-   * queueName\t  // app2
-   * </pre>
-   */
-  private void mockApplications(String appsConfig) {
-    int id = 1;
-    for (String a : appsConfig.split(";")) {
-      String[] strs = a.split("\t");
-      String queueName = strs[0];
-
-      // get containers
-      List<RMContainer> liveContainers = new ArrayList<RMContainer>();
-      List<RMContainer> reservedContainers = new ArrayList<RMContainer>();
-      ApplicationId appId = ApplicationId.newInstance(0L, id);
-      ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId, 1);
-
-      mockContainers(strs[1], appAttemptId, queueName, reservedContainers,
-          liveContainers);
-
-      FiCaSchedulerApp app = mock(FiCaSchedulerApp.class);
-      when(app.getLiveContainers()).thenReturn(liveContainers);
-      when(app.getReservedContainers()).thenReturn(reservedContainers);
-      when(app.getApplicationAttemptId()).thenReturn(appAttemptId);
-      when(app.getApplicationId()).thenReturn(appId);
-      when(app.getPriority()).thenReturn(Priority.newInstance(0));
-
-      // add to LeafQueue
-      LeafQueue queue = (LeafQueue) nameToCSQueues.get(queueName);
-      queue.getApplications().add(app);
-
-      id++;
-    }
-  }
-
-  /**
-   * Format is:
-   * host1=partition;
-   * host2=partition;
-   */
-  private void mockSchedulerNodes(String schedulerNodesConfigStr)
-      throws IOException {
-    String[] nodesConfigStrArray = schedulerNodesConfigStr.split(";");
-    for (String p : nodesConfigStrArray) {
-      NodeId nodeId = NodeId.newInstance(p.substring(0, p.indexOf("=")), 1);
-      String partition = p.substring(p.indexOf("=") + 1, p.length());
-
-      SchedulerNode sn = mock(SchedulerNode.class);
-      when(sn.getNodeID()).thenReturn(nodeId);
-      when(sn.getPartition()).thenReturn(partition);
-      nodeIdToSchedulerNodes.put(nodeId, sn);
-
-      LOG.debug("add scheduler node, id=" + nodeId + ", partition=" + partition);
-    }
-  }
-
-  /**
-   * Format is:
-   * <pre>
-   * partition0=total_resource,exclusivity;
-   * partition1=total_resource,exclusivity;
-   * ...
-   * </pre>
-   */
-  private void mockNodeLabelsManager(String nodeLabelsConfigStr) throws IOException {
-    String[] partitionConfigArr = nodeLabelsConfigStr.split(";");
-    clusterResource = Resources.createResource(0);
-    for (String p : partitionConfigArr) {
-      String partitionName = p.substring(0, p.indexOf("="));
-      Resource res = parseResourceFromString(p.substring(p.indexOf("=") + 1,
-          p.indexOf(",")));
-     boolean exclusivity =
-          Boolean.parseBoolean(p.substring(p.indexOf(",") + 1, p.length()));
-      when(nlm.getResourceByLabel(eq(partitionName), any(Resource.class)))
-          .thenReturn(res);
-      when(nlm.isExclusiveNodeLabel(eq(partitionName))).thenReturn(exclusivity);
-
-      // add to partition to resource
-      partitionToResource.put(partitionName, res);
-      LOG.debug("add partition=" + partitionName + " totalRes=" + res
-          + " exclusivity=" + exclusivity);
-      Resources.addTo(clusterResource, res);
-    }
-
-    when(nlm.getClusterNodeLabelNames()).thenReturn(
-        partitionToResource.keySet());
-  }
-
-  private Resource parseResourceFromString(String p) {
-    String[] resource = p.split(":");
-    Resource res = Resources.createResource(0);
-    if (resource.length == 1) {
-      res = Resources.createResource(Integer.parseInt(resource[0]));
-    } else {
-      res = Resources.createResource(Integer.parseInt(resource[0]),
-          Integer.parseInt(resource[1]));
-    }
-    return res;
-  }
-
-  /**
-   * Format is:
-   * <pre>
-   * root (<partition-name-1>=[guaranteed max used pending],<partition-name-2>=..);
-   * -A(...);
-   * --A1(...);
-   * --A2(...);
-   * -B...
-   * </pre>
-   * ";" splits queues, and there should no empty lines, no extra spaces
-   */
-  @SuppressWarnings({ "unchecked", "rawtypes" })
-  private ParentQueue mockQueueHierarchy(String queueExprs) {
-    String[] queueExprArray = queueExprs.split(";");
-    ParentQueue rootQueue = null;
-    for (int idx = 0; idx < queueExprArray.length; idx++) {
-      String q = queueExprArray[idx];
-      CSQueue queue;
-
-      // Initialize queue
-      if (isParent(queueExprArray, idx)) {
-        ParentQueue parentQueue = mock(ParentQueue.class);
-        queue = parentQueue;
-        List<CSQueue> children = new ArrayList<CSQueue>();
-        when(parentQueue.getChildQueues()).thenReturn(children);
-      } else {
-        LeafQueue leafQueue = mock(LeafQueue.class);
-        final TreeSet<FiCaSchedulerApp> apps = new TreeSet<>(
-            new Comparator<FiCaSchedulerApp>() {
-              @Override
-              public int compare(FiCaSchedulerApp a1, FiCaSchedulerApp a2) {
-                return a1.getApplicationId().compareTo(a2.getApplicationId());
-              }
-            });
-        when(leafQueue.getApplications()).thenReturn(apps);
-        OrderingPolicy<FiCaSchedulerApp> so = mock(OrderingPolicy.class);
-        when(so.getPreemptionIterator()).thenAnswer(new Answer() {
-          public Object answer(InvocationOnMock invocation) {
-            return apps.descendingIterator();
-          }
-        });
-        when(leafQueue.getOrderingPolicy()).thenReturn(so);
-
-        Map<String, TreeSet<RMContainer>> ignorePartitionContainers =
-            new HashMap<>();
-        when(leafQueue.getIgnoreExclusivityRMContainers()).thenReturn(
-            ignorePartitionContainers);
-        queue = leafQueue;
-      }
-
-      setupQueue(queue, q, queueExprArray, idx);
-      if (queue.getQueueName().equals(ROOT)) {
-        rootQueue = (ParentQueue) queue;
-      }
-    }
-    return rootQueue;
-  }
-
-  private void setupQueue(CSQueue queue, String q, String[] queueExprArray,
-      int idx) {
-    LOG.debug("*** Setup queue, source=" + q);
-    String queuePath = null;
-
-    int myLevel = getLevel(q);
-    if (0 == myLevel) {
-      // It's root
-      when(queue.getQueueName()).thenReturn(ROOT);
-      queuePath = ROOT;
-    }
-
-    String queueName = getQueueName(q);
-    when(queue.getQueueName()).thenReturn(queueName);
-
-    // Setup parent queue, and add myself to parentQueue.children-list
-    ParentQueue parentQueue = getParentQueue(queueExprArray, idx, myLevel);
-    if (null != parentQueue) {
-      when(queue.getParent()).thenReturn(parentQueue);
-      parentQueue.getChildQueues().add(queue);
-
-      // Setup my path
-      queuePath = parentQueue.getQueuePath() + "." + queueName;
-    }
-    when(queue.getQueuePath()).thenReturn(queuePath);
-
-    QueueCapacities qc = new QueueCapacities(0 == myLevel);
-    ResourceUsage ru = new ResourceUsage();
-
-    when(queue.getQueueCapacities()).thenReturn(qc);
-    when(queue.getQueueResourceUsage()).thenReturn(ru);
-
-    LOG.debug("Setup queue, name=" + queue.getQueueName() + " path="
-        + queue.getQueuePath());
-    LOG.debug("Parent=" + (parentQueue == null ? "null" : parentQueue
-        .getQueueName()));
-
-    // Setup other fields like used resource, guaranteed resource, etc.
-    String capacitySettingStr = q.substring(q.indexOf("(") + 1, q.indexOf(")"));
-    for (String s : capacitySettingStr.split(",")) {
-      String partitionName = s.substring(0, s.indexOf("="));
-      String[] values = s.substring(s.indexOf("[") + 1, s.indexOf("]")).split(" ");
-      // Add a small epsilon to capacities to avoid truncate when doing
-      // Resources.multiply
-      float epsilon = 1e-6f;
-      Resource totResoucePerPartition = partitionToResource.get(partitionName);
-      float absGuaranteed = Resources.divide(rc, totResoucePerPartition,
-          parseResourceFromString(values[0].trim()), totResoucePerPartition)
-          + epsilon;
-      float absMax = Resources.divide(rc, totResoucePerPartition,
-          parseResourceFromString(values[1].trim()), totResoucePerPartition)
-          + epsilon;
-      float absUsed = Resources.divide(rc, totResoucePerPartition,
-          parseResourceFromString(values[2].trim()), totResoucePerPartition)
-          + epsilon;
-      Resource pending = parseResourceFromString(values[3].trim());
-      qc.setAbsoluteCapacity(partitionName, absGuaranteed);
-      qc.setAbsoluteMaximumCapacity(partitionName, absMax);
-      qc.setAbsoluteUsedCapacity(partitionName, absUsed);
-      ru.setPending(partitionName, pending);
-      if (!isParent(queueExprArray, idx)) {
-        LeafQueue lq = (LeafQueue) queue;
-        when(lq.getTotalPendingResourcesConsideringUserLimit(isA(Resource.class),
-            isA(String.class))).thenReturn(pending);
-      }
-      ru.setUsed(partitionName, parseResourceFromString(values[2].trim()));
-      LOG.debug("Setup queue=" + queueName + " partition=" + partitionName
-          + " [abs_guaranteed=" + absGuaranteed + ",abs_max=" + absMax
-          + ",abs_used" + absUsed + ",pending_resource=" + pending + "]");
-    }
-
-    // Setup preemption disabled
-    when(queue.getPreemptionDisabled()).thenReturn(
-        conf.getPreemptionDisabled(queuePath, false));
-
-    nameToCSQueues.put(queueName, queue);
-    when(cs.getQueue(eq(queueName))).thenReturn(queue);
-  }
-
-  /**
-   * Level of a queue is how many "-" at beginning, root's level is 0
-   */
-  private int getLevel(String q) {
-    int level = 0; // level = how many "-" at beginning
-    while (level < q.length() && q.charAt(level) == '-') {
-      level++;
-    }
-    return level;
-  }
-
-  private String getQueueName(String q) {
-    int idx = 0;
-    // find first != '-' char
-    while (idx < q.length() && q.charAt(idx) == '-') {
-      idx++;
-    }
-    if (idx == q.length()) {
-      throw new IllegalArgumentException("illegal input:" + q);
-    }
-    // name = after '-' and before '('
-    String name = q.substring(idx, q.indexOf('('));
-    if (name.isEmpty()) {
-      throw new IllegalArgumentException("queue name shouldn't be empty:" + q);
-    }
-    if (name.contains(".")) {
-      throw new IllegalArgumentException("queue name shouldn't contain '.':"
-          + name);
-    }
-    return name;
-  }
-
-  private ParentQueue getParentQueue(String[] queueExprArray, int idx, int myLevel) {
-    idx--;
-    while (idx >= 0) {
-      int level = getLevel(queueExprArray[idx]);
-      if (level < myLevel) {
-        String parentQueuName = getQueueName(queueExprArray[idx]);
-        return (ParentQueue) nameToCSQueues.get(parentQueuName);
-      }
-      idx--;
-    }
-
-    return null;
-  }
-
-  /**
-   * Get if a queue is ParentQueue
-   */
-  private boolean isParent(String[] queues, int idx) {
-    int myLevel = getLevel(queues[idx]);
-    idx++;
-    while (idx < queues.length && getLevel(queues[idx]) == myLevel) {
-      idx++;
-    }
-    if (idx >= queues.length || getLevel(queues[idx]) < myLevel) {
-      // It's a LeafQueue
-      return false;
-    } else {
-      return true;
-    }
-  }
 }

http://git-wip-us.apache.org/repos/asf/hadoop/blob/bb62e059/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForReservedContainers.java
----------------------------------------------------------------------
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForReservedContainers.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForReservedContainers.java
new file mode 100644
index 0000000..38b2e78
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/monitor/capacity/TestProportionalCapacityPreemptionPolicyForReservedContainers.java
@@ -0,0 +1,430 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.monitor.capacity;
+
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+
+import static org.mockito.Matchers.argThat;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class TestProportionalCapacityPreemptionPolicyForReservedContainers
+    extends ProportionalCapacityPreemptionPolicyMockFramework {
+  @Before
+  public void setup() {
+    super.setup();
+    conf.setBoolean(
+        CapacitySchedulerConfiguration.PREEMPTION_SELECT_CANDIDATES_FOR_RESERVED_CONTAINERS,
+        true);
+    policy = new ProportionalCapacityPreemptionPolicy(rmContext, cs, mClock);
+  }
+
+  @Test
+  public void testPreemptionForSimpleReservedContainer() throws IOException {
+    /**
+     * The simplest test of reserved container, Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     * Guaranteed resource of a/b are 50:50
+     * Total cluster resource = 100
+     * - A has 90 containers on two node, n1 has 45, n2 has 45, size of each
+     * container is 1.
+     * - B has am container at n1, and reserves 1 container with size = 9 at n1,
+     * so B needs to preempt 9 containers from A at n1 instead of randomly
+     * preempt from n1 and n2.
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig = // n1 / n2 has no label
+        "n1= res=50;" +
+        "n2= res=50";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 100 9 9]);" + //root
+            "-a(=[50 100 90 0]);" + // a
+            "-b(=[50 100 10 9 9])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,1,n1,,45,false)" // 45 in n1
+            + "(1,1,n2,,45,false);" + // 45 in n2
+        "b\t" // app2 in b
+            + "(1,1,n1,,1,false)" // AM container in n1
+            + "(1,9,n1,,1,true)"; // 1 container with size=9 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    // Total 5 preempted from app1 at n1, don't preempt container from other
+    // app/node
+    verify(mDisp, times(5)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(1))));
+    verify(mDisp, times(5)).handle(
+        argThat(new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n1", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(2))));
+  }
+
+  @Test
+  public void testUseReservedAndFifoSelectorTogether() throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     * Guaranteed resource of a/b are 30:70
+     * Total cluster resource = 100
+     * - A has 45 containers on two node, n1 has 10, n2 has 35, size of each
+     * container is 1.
+     * - B has 5 containers at n2, and reserves 1 container with size = 50 at n1,
+     *   B also has 20 pending resources.
+     * so B needs to preempt:
+     * - 10 containers from n1 (for reserved)
+     * - 5 containers from n2 for pending resources
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig = // n1 / n2 has no label
+        "n1= res=50;" +
+        "n2= res=50";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 100 70 10]);" + //root
+            "-a(=[30 100 45 0]);" + // a
+            "-b(=[70 100 55 70 50])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,1,n2,,35,false)" // 35 in n2
+            + "(1,1,n1,,10,false);" + // 10 in n1
+            "b\t" // app2 in b
+            + "(1,1,n2,,5,false)" // 5 in n2
+            + "(1,50,n1,,1,true)"; // 1 container with size=50 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    verify(mDisp, times(15)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(1))));
+    verify(mDisp, times(10)).handle(
+        argThat(new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n1", 1))));
+    verify(mDisp, times(5)).handle(
+        argThat(new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n2", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(2))));
+  }
+
+  @Test
+  public void testReservedSelectorSkipsAMContainer() throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     * Guaranteed resource of a/b are 30:70
+     * Total cluster resource = 100
+     * - A has 45 containers on two node, n1 has 10, n2 has 35, size of each
+     * container is 1.
+     * - B has 5 containers at n2, and reserves 1 container with size = 50 at n1,
+     *   B also has 20 pending resources.
+     *
+     * Ideally B needs to preempt:
+     * - 10 containers from n1 (for reserved)
+     * - 5 containers from n2 for pending resources
+     *
+     * However, since one AM container is located at n1 (from queueA), we cannot
+     * preempt 10 containers from n1 for reserved container. Instead, we will
+     * preempt 15 containers from n2, since containers from queueA launched in n2
+     * are later than containers from queueA launched in n1 (FIFO order of containers)
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig = // n1 / n2 has no label
+        "n1= res=50;" +
+            "n2= res=50";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 100 70 10]);" + //root
+            "-a(=[30 100 45 0]);" + // a
+            "-b(=[70 100 55 70 50])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,1,n1,,10,false)" // 10 in n1
+            + "(1,1,n2,,35,false);" +// 35 in n2
+            "b\t" // app2 in b
+            + "(1,1,n2,,5,false)" // 5 in n2
+            + "(1,50,n1,,1,true)"; // 1 container with size=50 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    verify(mDisp, times(15)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(1))));
+    verify(mDisp, times(0)).handle(
+        argThat(new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n1", 1))));
+    verify(mDisp, times(15)).handle(
+        argThat(new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n2", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(2))));
+  }
+
+  @Test
+  public void testPreemptionForReservedContainerRespectGuaranteedResource()
+      throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     * Guaranteed resource of a/b are 85:15
+     * Total cluster resource = 100
+     * - A has 90 containers on two node, n1 has 45, n2 has 45, size of each
+     * container is 1.
+     * - B has am container at n1, and reserves 1 container with size = 9 at n1,
+     *
+     * If we preempt 9 containers from queue-A, queue-A will be below its
+     * guaranteed resource = 90 - 9 = 81 < 85.
+     *
+     * So no preemption will take place
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig = // n1 / n2 has no label
+        "n1= res=50;" +
+            "n2= res=50";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 100 9 9]);" + //root
+            "-a(=[85 100 90 0]);" + // a
+            "-b(=[15 100 10 9 9])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,1,n1,,45,false)" // 45 in n1
+            + "(1,1,n2,,45,false);" + // 45 in n2
+            "b\t" // app2 in b
+            + "(1,1,n1,,1,false)" // AM container in n1
+            + "(1,9,n1,,1,true)"; // 1 container with size=9 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    verify(mDisp, times(0)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new TestProportionalCapacityPreemptionPolicy.IsPreemptionRequestFor(
+            getAppAttemptId(2))));
+  }
+
+  @Test
+  public void testPreemptionForReservedContainerWhichHasAvailableResource()
+      throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     *
+     * Guaranteed resource of a/b are 50:50
+     * Total cluster resource = 100
+     * - A has 90 containers on two node, n1 has 45, n2 has 45, size of each
+     * container is 1.
+     * - B has am container at n1, and reserves 1 container with size = 9 at n1,
+     *
+     * So we can get 4 containers preempted after preemption.
+     * (reserved 5 + preempted 4) = 9
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig = // n1 / n2 has no label
+        "n1= res=50;" +
+            "n2= res=50";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 99 9 9]);" + //root
+            "-a(=[50 100 90 0]);" + // a
+            "-b(=[50 100 9 9 9])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,1,n1,,45,false)" // 45 in n1
+            + "(1,1,n2,,45,false);" + // 45 in n2
+            "b\t" // app2 in b
+            + "(1,9,n1,,1,true)"; // 1 container with size=9 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    // Total 4 preempted from app1 at n1, don't preempt container from other
+    // app/node
+    verify(mDisp, times(4)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n1", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n2", 1))));
+  }
+
+  @Test
+  public void testPreemptionForReservedContainerWhichHasNondivisibleAvailableResource()
+      throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     *
+     * Guaranteed resource of a/b are 50:50
+     * Total cluster resource = 100
+     * - A has 45 containers on two node, size of each container is 2,
+     *   n1 has 23, n2 has 22
+     * - B reserves 1 container with size = 9 at n1,
+     *
+     * So we can get 4 containers (total-resource = 8) preempted after
+     * preemption. Actual required is 3.5, but we need to preempt integer
+     * number of containers
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig = // n1 / n2 has no label
+        "n1= res=50;" +
+            "n2= res=50";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 99 9 9]);" + //root
+            "-a(=[50 100 90 0]);" + // a
+            "-b(=[50 100 9 9 9])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,2,n1,,24,false)" // 48 in n1
+            + "(1,2,n2,,23,false);" + // 46 in n2
+            "b\t" // app2 in b
+            + "(1,9,n1,,1,true)"; // 1 container with size=9 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    // Total 4 preempted from app1 at n1, don't preempt container from other
+    // app/node
+    verify(mDisp, times(4)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n1", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n2", 1))));
+  }
+
+  @Test
+  public void testPreemptionForReservedContainerRespectAvailableResources()
+      throws IOException {
+    /**
+     * Queue structure is:
+     *
+     * <pre>
+     *       root
+     *       /  \
+     *      a    b
+     * </pre>
+     *
+     * Guaranteed resource of a/b are 50:50
+     * Total cluster resource = 100, 4 nodes, 25 on each node
+     * - A has 10 containers on every node, size of container is 2
+     * - B reserves 1 container with size = 9 at n1,
+     *
+     * So even if we cannot allocate container for B now, no preemption should
+     * happen since there're plenty of available resources.
+     */
+    String labelsConfig =
+        "=100,true;";
+    String nodesConfig =
+        "n1= res=25;" +
+            "n2= res=25;" +
+            "n3= res=25;" +
+            "n4= res=25;";
+    String queuesConfig =
+        // guaranteed,max,used,pending,reserved
+        "root(=[100 100 89 9 9]);" + //root
+            "-a(=[50 100 80 0]);" + // a
+            "-b(=[50 100 9 9 9])"; // b
+    String appsConfig=
+        //queueName\t(priority,resource,host,expression,#repeat,reserved)
+        "a\t" // app1 in a
+            + "(1,2,n1,,10,false)" // 10 in n1
+            + "(1,2,n2,,10,false)" // 10 in n2
+            + "(1,2,n3,,10,false)" // 10 in n3
+            + "(1,2,n4,,10,false);" + // 10 in n4
+            "b\t" // app2 in b
+            + "(1,9,n1,,1,true)"; // 1 container with size=5 reserved at n1
+
+    buildEnv(labelsConfig, nodesConfig, queuesConfig, appsConfig);
+    policy.editSchedule();
+
+    // No preemption should happen
+    verify(mDisp, times(0)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n1", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n2", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n3", 1))));
+    verify(mDisp, times(0)).handle(argThat(
+        new IsPreemptionRequestForQueueAndNode(getAppAttemptId(1), "a",
+            NodeId.newInstance("n4", 1))));
+  }
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message