/**
 * 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.scheduler.capacity;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.when;

import java.io.IOException;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.QueueACL;
import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.security.YarnAuthorizationProvider;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.nodelabels.RMNodeLabelsManager;
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.capacity.preemption.PreemptionManager;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ContainerAllocationProposal;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.ResourceCommitRequest;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.common.SchedulerContainer;
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.placement.CandidateNodeSet;
import org.apache.hadoop.yarn.server.resourcemanager.security.AppPriorityACLsManager;
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.InOrder;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;

public class TestParentQueue {

  private static final Resource QUEUE_B_RESOURCE = Resource
      .newInstance(14 * 1024, 22);
  private static final Resource QUEUE_A_RESOURCE = Resource
      .newInstance(6 * 1024, 10);

  private static final Logger LOG =
      LoggerFactory.getLogger(TestParentQueue.class);
  
  RMContext rmContext;
  YarnConfiguration conf;
  CapacitySchedulerConfiguration csConf;
  CapacitySchedulerContext csContext;
  
  final static int GB = 1024;
  final static String DEFAULT_RACK = "/default";

  private final ResourceCalculator resourceComparator =
      new DefaultResourceCalculator();
  
  @Before
  public void setUp() throws Exception {
    rmContext = TestUtils.getMockRMContext();
    conf = new YarnConfiguration();
    csConf = new CapacitySchedulerConfiguration();
    
    csContext = mock(CapacitySchedulerContext.class);
    when(csContext.getConf()).thenReturn(conf);
    when(csContext.getConfiguration()).thenReturn(csConf);
    when(csContext.getMinimumResourceCapability()).thenReturn(
        Resources.createResource(GB, 1));
    when(csContext.getMaximumResourceCapability()).thenReturn(
        Resources.createResource(16*GB, 32));
    when(csContext.getClusterResource()).
        thenReturn(Resources.createResource(100 * 16 * GB, 100 * 32));
    when(csContext.getPreemptionManager()).thenReturn(new PreemptionManager());
    when(csContext.getResourceCalculator()).
        thenReturn(resourceComparator);
    when(csContext.getRMContext()).thenReturn(rmContext);
  }
  
  private static final String A = "a";
  private static final String B = "b";
  private static final String Q_A =
      CapacitySchedulerConfiguration.ROOT + "." + A;
  private static final String Q_B =
      CapacitySchedulerConfiguration.ROOT + "." + B;
  private void setupSingleLevelQueues(CapacitySchedulerConfiguration conf) {
    
    // Define top-level queues
    conf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {A, B});
    
    conf.setCapacity(Q_A, 30);
    
    conf.setCapacity(Q_B, 70);
    
    LOG.info("Setup top-level queues a and b");
  }

  private void setupSingleLevelQueuesWithAbsoluteResource(
      CapacitySchedulerConfiguration conf) {

    // Define top-level queues
    conf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[]{A, B});

    conf.setMinimumResourceRequirement("", Q_A,
        QUEUE_A_RESOURCE);

    conf.setMinimumResourceRequirement("", Q_B,
        QUEUE_B_RESOURCE);

    LOG.info("Setup top-level queues a and b with absolute resource");
  }

  private FiCaSchedulerApp getMockApplication(int appId, String user) {
    FiCaSchedulerApp application = mock(FiCaSchedulerApp.class);
    doReturn(user).when(application).getUser();
    doReturn(Resources.createResource(0, 0)).when(application).getHeadroom();
    return application;
  }

  private void applyAllocationToQueue(Resource clusterResource,
      int allocatedMem,
      CSQueue queue) {
    // Call accept & apply for queue
    ResourceCommitRequest request = mock(ResourceCommitRequest.class);
    when(request.anythingAllocatedOrReserved()).thenReturn(true);
    ContainerAllocationProposal allocation = mock(
        ContainerAllocationProposal.class);
    when(request.getTotalReleasedResource()).thenReturn(Resources.none());
    when(request.getFirstAllocatedOrReservedContainer()).thenReturn(allocation);
    SchedulerContainer scontainer = mock(SchedulerContainer.class);
    when(allocation.getAllocatedOrReservedContainer()).thenReturn(scontainer);
    when(allocation.getAllocatedOrReservedResource()).thenReturn(
        Resources.createResource(allocatedMem));
    when(scontainer.getNodePartition()).thenReturn("");

    if (queue.accept(clusterResource, request)) {
      queue.apply(clusterResource, request);
    }
  }

  private void stubQueueAllocation(final CSQueue queue, 
      final Resource clusterResource, final FiCaSchedulerNode node, 
      final int allocation) {
    stubQueueAllocation(queue, clusterResource, node, allocation, 
        NodeType.NODE_LOCAL);
  }
  
  private void stubQueueAllocation(final CSQueue queue, 
      final Resource clusterResource, final FiCaSchedulerNode node, 
      final int allocation, final NodeType type) {
    
    // Simulate the queue allocation
    doAnswer(new Answer<CSAssignment>() {
      @Override
      public CSAssignment answer(InvocationOnMock invocation) throws Throwable {
        try {
          throw new Exception();
        } catch (Exception e) {
          LOG.info("FOOBAR q.assignContainers q=" + queue.getQueuePath() +
              " alloc=" + allocation + " node=" + node.getNodeName());
        }
        final Resource allocatedResource = Resources.createResource(allocation);
        if (queue instanceof ParentQueue) {
          ((ParentQueue)queue).allocateResource(clusterResource, 
              allocatedResource, RMNodeLabelsManager.NO_LABEL);
        } else {
          FiCaSchedulerApp app1 = getMockApplication(0, "");
          ((LeafQueue)queue).allocateResource(clusterResource, app1, 
              allocatedResource, null, null);
        }
        
        // Next call - nothing
        if (allocation > 0) {
          doReturn(new CSAssignment(Resources.none(), type)).when(queue)
              .assignContainers(eq(clusterResource),
                  any(CandidateNodeSet.class), any(ResourceLimits.class),
                  any(SchedulingMode.class));

          // Mock the node's resource availability
          Resource available = node.getUnallocatedResource();
          doReturn(Resources.subtractFrom(available, allocatedResource)).
          when(node).getUnallocatedResource();
        }

        return new CSAssignment(allocatedResource, type);
      }
    }).when(queue).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), any(ResourceLimits.class),
        any(SchedulingMode.class));
  }
  
  private float computeQueueAbsoluteUsedCapacity(CSQueue queue, 
      int expectedMemory, Resource clusterResource) {
    return (
        ((float)expectedMemory / (float)clusterResource.getMemorySize())
      );
  }
  
  private float computeQueueUsedCapacity(CSQueue queue,
      int expectedMemory, Resource clusterResource) {
    return (expectedMemory / 
        (clusterResource.getMemorySize() * queue.getAbsoluteCapacity()));
  }
  
  final static float DELTA = 0.0001f;
  private void verifyQueueMetrics(CSQueue queue, 
      int expectedMemory, Resource clusterResource) {
    assertEquals(
        computeQueueAbsoluteUsedCapacity(queue, expectedMemory, clusterResource), 
        queue.getAbsoluteUsedCapacity(), 
        DELTA);
    assertEquals(
        computeQueueUsedCapacity(queue, expectedMemory, clusterResource), 
        queue.getUsedCapacity(), 
        DELTA);

  }
  
  @Test
  public void testSingleLevelQueues() throws Exception {
    // Setup queue configs
    setupSingleLevelQueues(csConf);

    CSQueueStore queues = new CSQueueStore();
    CSQueue root =
        CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
            CapacitySchedulerConfiguration.ROOT, queues, queues, 
            TestUtils.spyHook);

    // Setup some nodes
    final int memoryPerNode = 10;
    final int coresPerNode = 16;
    final int numNodes = 2;
    
    FiCaSchedulerNode node_0 = 
        TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB);
    FiCaSchedulerNode node_1 = 
        TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB);
    
    final Resource clusterResource = 
        Resources.createResource(numNodes * (memoryPerNode*GB),
            numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    // Start testing
    LeafQueue a = (LeafQueue)queues.get(A);
    LeafQueue b = (LeafQueue)queues.get(B);
    
    a.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage()
    .incPending(Resources.createResource(1 * GB));
    
    // Simulate B returning a container on node_0
    stubQueueAllocation(a, clusterResource, node_0, 0*GB);
    stubQueueAllocation(b, clusterResource, node_0, 1*GB);
    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    verifyQueueMetrics(a, 0*GB, clusterResource);
    verifyQueueMetrics(b, 1*GB, clusterResource);
    
    // Now, A should get the scheduling opportunity since A=0G/6G, B=1G/14G
    stubQueueAllocation(a, clusterResource, node_1, 2*GB);
    stubQueueAllocation(b, clusterResource, node_1, 1*GB);
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    InOrder allocationOrder = inOrder(a, b);
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 2*GB, clusterResource);
    verifyQueueMetrics(b, 2*GB, clusterResource);

    // Now, B should get the scheduling opportunity 
    // since A has 2/6G while B has 2/14G
    stubQueueAllocation(a, clusterResource, node_0, 1*GB);
    stubQueueAllocation(b, clusterResource, node_0, 2*GB);
    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    root.assignContainers(clusterResource, node_0,
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(b, a);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 3*GB, clusterResource);
    verifyQueueMetrics(b, 4*GB, clusterResource);

    // Now, B should still get the scheduling opportunity 
    // since A has 3/6G while B has 4/14G
    stubQueueAllocation(a, clusterResource, node_0, 0*GB);
    stubQueueAllocation(b, clusterResource, node_0, 4*GB);
    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(b, a);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 3*GB, clusterResource);
    verifyQueueMetrics(b, 8*GB, clusterResource);

    // Now, A should get the scheduling opportunity 
    // since A has 3/6G while B has 8/14G
    stubQueueAllocation(a, clusterResource, node_1, 1*GB);
    stubQueueAllocation(b, clusterResource, node_1, 1*GB);
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(a, b);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 4*GB, clusterResource);
    verifyQueueMetrics(b, 9*GB, clusterResource);
  }

  @Test
  public void testSingleLevelQueuesPrecision() throws Exception {
    // Setup queue configs
    setupSingleLevelQueues(csConf);
    csConf.setCapacity(Q_A, 30);
    csConf.setCapacity(Q_B, 70.5F);

    CSQueueStore queues = new CSQueueStore();
    boolean exceptionOccurred = false;
    try {
      CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
          CapacitySchedulerConfiguration.ROOT, queues, queues,
          TestUtils.spyHook);
    } catch (IOException ie) {
      exceptionOccurred = true;
    }
    if (!exceptionOccurred) {
      Assert.fail("Capacity is more then 100% so should be failed.");
    }
    csConf.setCapacity(Q_A, 30);
    csConf.setCapacity(Q_B, 70);
    exceptionOccurred = false;
    queues.clear();
    try {
      CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
          CapacitySchedulerConfiguration.ROOT, queues, queues,
          TestUtils.spyHook);
    } catch (IllegalArgumentException ie) {
      exceptionOccurred = true;
    }
    if (exceptionOccurred) {
      Assert.fail("Capacity is 100% so should not be failed.");
    }
    csConf.setCapacity(Q_A, 30);
    csConf.setCapacity(Q_B, 70.005F);
    exceptionOccurred = false;
    queues.clear();
    try {
      CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
          CapacitySchedulerConfiguration.ROOT, queues, queues,
          TestUtils.spyHook);
    } catch (IllegalArgumentException ie) {
      exceptionOccurred = true;
    }
    if (exceptionOccurred) {
      Assert
          .fail("Capacity is under PRECISION which is .05% so should not be failed.");
    }
  }
  
  private static final String C = "c";
  private static final String C1 = "c1";
  private static final String C11 = "c11";
  private static final String C111 = "c111";
  private static final String C1111 = "c1111";

  private static final String D = "d";
  private static final String A1 = "a1";
  private static final String A2 = "a2";
  private static final String B1 = "b1";
  private static final String B2 = "b2";
  private static final String B3 = "b3";
  private static final String B4 = "b4";
  
  private void setupMultiLevelQueues(CapacitySchedulerConfiguration conf) {
    
    // Define top-level queues
    csConf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {A, B, C, D});
    
    conf.setCapacity(Q_A, 10);
    
    conf.setCapacity(Q_B, 50);
    
    final String Q_C = CapacitySchedulerConfiguration.ROOT + "." + C;
    conf.setCapacity(Q_C, 19.5f);
    
    final String Q_D = CapacitySchedulerConfiguration.ROOT + "." + D;
    conf.setCapacity(Q_D, 20.5f);
    
    // Define 2-nd level queues
    conf.setQueues(Q_A, new String[] {A1, A2});
    conf.setCapacity(Q_A + "." + A1, 50);
    conf.setCapacity(Q_A + "." + A2, 50);
    
    conf.setQueues(Q_B, new String[] {B1, B2, B3});
    conf.setCapacity(Q_B + "." + B1, 10);
    conf.setCapacity(Q_B + "." + B2, 20);
    conf.setCapacity(Q_B + "." + B3, 70);

    conf.setQueues(Q_C, new String[] {C1});

    final String Q_C1= Q_C + "." + C1;
    conf.setCapacity(Q_C1, 100);
    conf.setQueues(Q_C1, new String[] {C11});

    final String Q_C11= Q_C1 + "." + C11;
    conf.setCapacity(Q_C11, 100);
    conf.setQueues(Q_C11, new String[] {C111});

    final String Q_C111= Q_C11 + "." + C111;
    conf.setCapacity(Q_C111, 100);
    //Leaf Queue
    conf.setQueues(Q_C111, new String[] {C1111});
    final String Q_C1111= Q_C111 + "." + C1111;
    conf.setCapacity(Q_C1111, 100);
  }

  @Test
  public void testMultiLevelQueues() throws Exception {
    /*
     * Structure of queue:
     *            Root
     *           ____________
     *          /    |   \   \
     *         A     B    C   D
     *       / |   / | \   \
     *      A1 A2 B1 B2 B3  C1
     *                        \
     *                         C11
     *                           \
     *                           C111
     *                             \
     *                              C1111
     */
    
    // Setup queue configs
    setupMultiLevelQueues(csConf);

    CSQueueStore queues = new CSQueueStore();
    CSQueue root =
        CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
            CapacitySchedulerConfiguration.ROOT, queues, queues,
            TestUtils.spyHook);
    
    // Setup some nodes
    final int memoryPerNode = 10;
    final int coresPerNode = 16;
    final int numNodes = 3;
    
    FiCaSchedulerNode node_0 = 
        TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB);
    FiCaSchedulerNode node_1 = 
        TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB);
    FiCaSchedulerNode node_2 = 
        TestUtils.getMockNode("host_2", DEFAULT_RACK, 0, memoryPerNode*GB);
    
    final Resource clusterResource = 
        Resources.createResource(numNodes * (memoryPerNode*GB), 
            numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    // Start testing
    CSQueue a = queues.get(A);
    a.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    CSQueue b = queues.get(B);
    b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    CSQueue c = queues.get(C);
    c.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    CSQueue d = queues.get(D);
    d.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));

    CSQueue a1 = queues.get(A1);
    a1.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    CSQueue a2 = queues.get(A2);
    a2.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));

    CSQueue b1 = queues.get(B1);
    b1.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    CSQueue b2 = queues.get(B2);
    b2.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    CSQueue b3 = queues.get(B3);
    b3.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage()
    .incPending(Resources.createResource(1 * GB));

    // Simulate C returning a container on node_0
    stubQueueAllocation(a, clusterResource, node_0, 0*GB);
    stubQueueAllocation(b, clusterResource, node_0, 0*GB);
    stubQueueAllocation(c, clusterResource, node_0, 1*GB);
    stubQueueAllocation(d, clusterResource, node_0, 0*GB);
    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    verifyQueueMetrics(a, 0*GB, clusterResource);
    verifyQueueMetrics(b, 0*GB, clusterResource);
    verifyQueueMetrics(c, 1*GB, clusterResource);
    verifyQueueMetrics(d, 0*GB, clusterResource);
    reset(a); reset(b); reset(c);

    // Now get B2 to allocate
    // A = 0/3, B = 0/15, C = 1/6, D=0/6
    stubQueueAllocation(a, clusterResource, node_1, 0*GB);
    stubQueueAllocation(b2, clusterResource, node_1, 4*GB);
    stubQueueAllocation(c, clusterResource, node_1, 0*GB);
    root.assignContainers(clusterResource, node_1, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    applyAllocationToQueue(clusterResource, 4*GB,
        b);
    verifyQueueMetrics(a, 0*GB, clusterResource);
    verifyQueueMetrics(b, 4*GB, clusterResource);
    verifyQueueMetrics(c, 1*GB, clusterResource);
    reset(a); reset(b); reset(c);
    
    // Now get both A1, C & B3 to allocate in right order
    // A = 0/3, B = 4/15, C = 1/6, D=0/6
    stubQueueAllocation(a1, clusterResource, node_0, 1*GB);
    stubQueueAllocation(b3, clusterResource, node_0, 2*GB);
    stubQueueAllocation(c, clusterResource, node_0, 2*GB);

    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    InOrder allocationOrder = inOrder(a, c, b);
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    applyAllocationToQueue(clusterResource, 1 * GB, a);

    root.assignContainers(clusterResource, node_0,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder.verify(c).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    applyAllocationToQueue(clusterResource, 2 * GB, root);

    root.assignContainers(clusterResource, node_0,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    applyAllocationToQueue(clusterResource, 2*GB, b);
    verifyQueueMetrics(a, 1*GB, clusterResource);
    verifyQueueMetrics(b, 6*GB, clusterResource);
    verifyQueueMetrics(c, 3*GB, clusterResource);
    reset(a); reset(b); reset(c);
    
    // Now verify max-capacity
    // A = 1/3, B = 6/15, C = 3/6, D=0/6
    // Ensure a1 won't alloc above max-cap although it should get 
    // scheduling opportunity now, right after a2
    LOG.info("here");
    ((ParentQueue)a).setMaxCapacity(.1f);  // a should be capped at 3/30
    stubQueueAllocation(a1, clusterResource, node_2, 1*GB); // shouldn't be 
                                                            // allocated due 
                                                            // to max-cap
    stubQueueAllocation(a2, clusterResource, node_2, 2*GB);
    stubQueueAllocation(b3, clusterResource, node_2, 1*GB);
    stubQueueAllocation(b1, clusterResource, node_2, 1*GB);
    stubQueueAllocation(c, clusterResource, node_2, 1*GB);
    root.assignContainers(clusterResource, node_2,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(a, a2, a1, b, c);
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(a2).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    applyAllocationToQueue(clusterResource, 2*GB, a);

    root.assignContainers(clusterResource, node_2,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    applyAllocationToQueue(clusterResource, 2*GB, b);

    root.assignContainers(clusterResource, node_2,
        new ResourceLimits(clusterResource),
        SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder.verify(c).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 3*GB, clusterResource);
    verifyQueueMetrics(b, 8*GB, clusterResource);
    verifyQueueMetrics(c, 4*GB, clusterResource);
    reset(a); reset(b); reset(c);
  }
  
  @Test (expected=IOException.class)
  public void testQueueCapacitySettingChildZero() throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);
    
    // set child queues capacity to 0 when parents not 0
    csConf.setCapacity(Q_B + "." + B1, 0);
    csConf.setCapacity(Q_B + "." + B2, 0);
    csConf.setCapacity(Q_B + "." + B3, 0);

    CSQueueStore queues = new CSQueueStore();
    CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
        CapacitySchedulerConfiguration.ROOT, queues, queues,
        TestUtils.spyHook);
  }
  
  @Test (expected=IOException.class)
  public void testQueueCapacitySettingParentZero() throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);
    
    // set parent capacity to 0 when child not 0
    csConf.setCapacity(Q_B, 0);
    csConf.setCapacity(Q_A, 60);

    CSQueueStore queues = new CSQueueStore();
    CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
        CapacitySchedulerConfiguration.ROOT, queues, queues,
        TestUtils.spyHook);
  }

  @Test
  public void testQueueCapacitySettingParentZeroChildren100pctZeroSumAllowed()
      throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);

    // set parent capacity to 0 when child is 100
    // and allow zero capacity sum
    csConf.setCapacity(Q_B, 0);
    csConf.setCapacity(Q_A, 60);
    csConf.setAllowZeroCapacitySum(Q_B, true);
    CSQueueStore queues = new CSQueueStore();
    CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
        CapacitySchedulerConfiguration.ROOT, queues, queues,
        TestUtils.spyHook);
  }

  @Test(expected = IOException.class)
  public void testQueueCapacitySettingParentZeroChildren50pctZeroSumAllowed()
      throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);

    // set parent capacity to 0 when sum(children) is 50
    // and allow zero capacity sum
    csConf.setCapacity(Q_B, 0);
    csConf.setCapacity(Q_A, 100);
    csConf.setCapacity(Q_B + "." + B1, 10);
    csConf.setCapacity(Q_B + "." + B2, 20);
    csConf.setCapacity(Q_B + "." + B3, 20);
    csConf.setAllowZeroCapacitySum(Q_B, true);
    CSQueueStore queues = new CSQueueStore();
    CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
        CapacitySchedulerConfiguration.ROOT, queues, queues,
        TestUtils.spyHook);
  }

  @Test
  public void testQueueCapacitySettingParentNonZeroChildrenZeroSumAllowed()
      throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);

    // set parent capacity to 10 when sum(children) is 0
    // and allow zero capacity sum
    csConf.setCapacity(Q_B, 10);
    csConf.setCapacity(Q_A, 50);
    csConf.setCapacity(Q_B + "." + B1, 0);
    csConf.setCapacity(Q_B + "." + B2, 0);
    csConf.setCapacity(Q_B + "." + B3, 0);
    csConf.setAllowZeroCapacitySum(Q_B, true);
    CSQueueStore queues = new CSQueueStore();
    CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
        CapacitySchedulerConfiguration.ROOT, queues, queues,
        TestUtils.spyHook);
  }

  @Test
  public void testQueueCapacityZero() throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);
    
    // set parent and child capacity to 0
    csConf.setCapacity(Q_B, 0);
    csConf.setCapacity(Q_B + "." + B1, 0);
    csConf.setCapacity(Q_B + "." + B2, 0);
    csConf.setCapacity(Q_B + "." + B3, 0);
    
    csConf.setCapacity(Q_A, 60);

    CSQueueStore queues = new CSQueueStore();
    try {
      CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
          CapacitySchedulerConfiguration.ROOT, queues, queues,
          TestUtils.spyHook);
    } catch (IllegalArgumentException e) {
      fail("Failed to create queues with 0 capacity: " + e);
    }
    assertTrue("Failed to create queues with 0 capacity", true);
  }

  @Test
  public void testOffSwitchScheduling() throws Exception {
    // Setup queue configs
    setupSingleLevelQueues(csConf);

    CSQueueStore queues = new CSQueueStore();
    CSQueue root =
        CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
            CapacitySchedulerConfiguration.ROOT, queues, queues,
            TestUtils.spyHook);

    // Setup some nodes
    final int memoryPerNode = 10;
    final int coresPerNode = 16;
    final int numNodes = 2;

    FiCaSchedulerNode node_0 =
        TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB);
    FiCaSchedulerNode node_1 =
        TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB);

    final Resource clusterResource =
        Resources.createResource(numNodes * (memoryPerNode*GB),
            numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    // Start testing
    LeafQueue a = (LeafQueue)queues.get(A);
    LeafQueue b = (LeafQueue)queues.get(B);
    a.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage()
        .incPending(Resources.createResource(1 * GB));

    // Simulate B returning a container on node_0
    stubQueueAllocation(a, clusterResource, node_0, 0*GB, NodeType.OFF_SWITCH);
    stubQueueAllocation(b, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH);
    root.assignContainers(clusterResource, node_0,
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    verifyQueueMetrics(a, 0*GB, clusterResource);
    verifyQueueMetrics(b, 1*GB, clusterResource);

    // Now, A should get the scheduling opportunity since A=0G/6G, B=1G/14G
    // also, B gets a scheduling opportunity since A allocates RACK_LOCAL
    stubQueueAllocation(a, clusterResource, node_1, 2*GB, NodeType.RACK_LOCAL);
    stubQueueAllocation(b, clusterResource, node_1, 1*GB, NodeType.OFF_SWITCH);
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    InOrder allocationOrder = inOrder(a);
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(b);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 2*GB, clusterResource);
    verifyQueueMetrics(b, 2*GB, clusterResource);

    // Now, B should get the scheduling opportunity
    // since A has 2/6G while B has 2/14G,
    // However, since B returns off-switch, A won't get an opportunity
    stubQueueAllocation(a, clusterResource, node_0, 1*GB, NodeType.NODE_LOCAL);
    stubQueueAllocation(b, clusterResource, node_0, 2*GB, NodeType.OFF_SWITCH);
    root.assignContainers(clusterResource, node_0,
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(b, a);
    allocationOrder.verify(b).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(a).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(a, 2*GB, clusterResource);
    verifyQueueMetrics(b, 4*GB, clusterResource);

  }
  
  @Test
  public void testOffSwitchSchedulingMultiLevelQueues() throws Exception {
    // Setup queue configs
    setupMultiLevelQueues(csConf);
    //B3
    CSQueueStore queues = new CSQueueStore();
    CSQueue root = 
        CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
            CapacitySchedulerConfiguration.ROOT, queues, queues,
            TestUtils.spyHook);

    // Setup some nodes
    final int memoryPerNode = 10;
    final int coresPerNode = 10;
    final int numNodes = 2;
    
    FiCaSchedulerNode node_0 = 
        TestUtils.getMockNode("host_0", DEFAULT_RACK, 0, memoryPerNode*GB);
    FiCaSchedulerNode node_1 = 
        TestUtils.getMockNode("host_1", DEFAULT_RACK, 0, memoryPerNode*GB);
    
    final Resource clusterResource = 
        Resources.createResource(numNodes * (memoryPerNode*GB),
            numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    // Start testing
    LeafQueue b3 = (LeafQueue)queues.get(B3);
    LeafQueue b2 = (LeafQueue)queues.get(B2);
    b2.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    b3.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    queues.get(CapacitySchedulerConfiguration.ROOT).getQueueResourceUsage()
    .incPending(Resources.createResource(1 * GB));
    
    CSQueue b = queues.get(B);
    b.getQueueResourceUsage().incPending(Resources.createResource(1 * GB));
    
    // Simulate B3 returning a container on node_0
    stubQueueAllocation(b2, clusterResource, node_0, 0*GB, NodeType.OFF_SWITCH);
    stubQueueAllocation(b3, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH);
    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    verifyQueueMetrics(b2, 0*GB, clusterResource);
    verifyQueueMetrics(b3, 1*GB, clusterResource);
    
    // Now, B2 should get the scheduling opportunity since B2=0G/2G, B3=1G/7G
    // also, B3 gets a scheduling opportunity since B2 allocates RACK_LOCAL
    stubQueueAllocation(b2, clusterResource, node_1, 1*GB, NodeType.RACK_LOCAL);
    stubQueueAllocation(b3, clusterResource, node_1, 1*GB, NodeType.OFF_SWITCH);
    root.assignContainers(clusterResource, node_1, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    root.assignContainers(clusterResource, node_1,
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    InOrder allocationOrder = inOrder(b2, b3);
    allocationOrder.verify(b2).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(b3).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(b2, 1*GB, clusterResource);
    verifyQueueMetrics(b3, 2*GB, clusterResource);
    
    // Now, B3 should get the scheduling opportunity 
    // since B2 has 1/2G while B3 has 2/7G, 
    // However, since B3 returns off-switch, B2 won't get an opportunity
    stubQueueAllocation(b2, clusterResource, node_0, 1*GB, NodeType.NODE_LOCAL);
    stubQueueAllocation(b3, clusterResource, node_0, 1*GB, NodeType.OFF_SWITCH);
    root.assignContainers(clusterResource, node_0, 
        new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY);
    allocationOrder = inOrder(b3, b2);
    allocationOrder.verify(b3).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    allocationOrder.verify(b2).assignContainers(eq(clusterResource),
        any(CandidateNodeSet.class), anyResourceLimits(),
        any(SchedulingMode.class));
    verifyQueueMetrics(b2, 1*GB, clusterResource);
    verifyQueueMetrics(b3, 3*GB, clusterResource);

  }

  public boolean hasQueueACL(List<QueueUserACLInfo> aclInfos, QueueACL acl, String qName) {
    for (QueueUserACLInfo aclInfo : aclInfos) {
      if (aclInfo.getQueueName().equals(qName)) {
        if (aclInfo.getUserAcls().contains(acl)) {
          return true;
        }
      }
    }    
    return false;
  }

  @Test
  public void testQueueAcl() throws Exception {
 
    setupMultiLevelQueues(csConf);
    csConf.setAcl(CapacitySchedulerConfiguration.ROOT, QueueACL.SUBMIT_APPLICATIONS, " ");
    csConf.setAcl(CapacitySchedulerConfiguration.ROOT, QueueACL.ADMINISTER_QUEUE, " ");

    final String Q_C = CapacitySchedulerConfiguration.ROOT + "." + C;
    csConf.setAcl(Q_C, QueueACL.ADMINISTER_QUEUE, "*");
    final String Q_C11= Q_C + "." + C1 +  "." + C11;
    csConf.setAcl(Q_C11, QueueACL.SUBMIT_APPLICATIONS, "*");

    CSQueueStore queues = new CSQueueStore();
    CSQueue root = 
        CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null,
            CapacitySchedulerConfiguration.ROOT, queues, queues,
            TestUtils.spyHook);
    YarnAuthorizationProvider authorizer =
        YarnAuthorizationProvider.getInstance(conf);
    AppPriorityACLsManager appPriorityACLManager = new AppPriorityACLsManager(
        conf);
    CapacitySchedulerQueueManager.setQueueAcls(authorizer,
        appPriorityACLManager, queues);

    UserGroupInformation user = UserGroupInformation.getCurrentUser();
    // Setup queue configs
    ParentQueue c = (ParentQueue)queues.get(C);
    ParentQueue c1 = (ParentQueue)queues.get(C1);
    ParentQueue c11 = (ParentQueue)queues.get(C11);
    ParentQueue c111 = (ParentQueue)queues.get(C111);

    assertFalse(root.hasAccess(QueueACL.ADMINISTER_QUEUE, user));
    List<QueueUserACLInfo> aclInfos = root.getQueueUserAclInfo(user);
    assertFalse(hasQueueACL(aclInfos, QueueACL.ADMINISTER_QUEUE, "root"));
    
    assertFalse(root.hasAccess(QueueACL.SUBMIT_APPLICATIONS, user));
    assertFalse(hasQueueACL(aclInfos, QueueACL.SUBMIT_APPLICATIONS, "root"));

    // c has no SA, but QA
    assertTrue(c.hasAccess(QueueACL.ADMINISTER_QUEUE, user));
    assertTrue(hasQueueACL(aclInfos,  QueueACL.ADMINISTER_QUEUE, "root.c"));
    assertFalse(c.hasAccess(QueueACL.SUBMIT_APPLICATIONS, user));
    assertFalse(hasQueueACL(aclInfos, QueueACL.SUBMIT_APPLICATIONS, "root.c"));

    //Queue c1 has QA, no SA (gotten perm from parent)
    assertTrue(c1.hasAccess(QueueACL.ADMINISTER_QUEUE, user)); 
    assertTrue(hasQueueACL(aclInfos,  QueueACL.ADMINISTER_QUEUE, "root.c.c1"));
    assertFalse(c1.hasAccess(QueueACL.SUBMIT_APPLICATIONS, user)); 
    assertFalse(hasQueueACL(
        aclInfos, QueueACL.SUBMIT_APPLICATIONS, "root.c.c1"));

    //Queue c11 has permissions from parent queue and SA
    assertTrue(c11.hasAccess(QueueACL.ADMINISTER_QUEUE, user));
    assertTrue(hasQueueACL(
        aclInfos,  QueueACL.ADMINISTER_QUEUE, "root.c.c1.c11"));
    assertTrue(c11.hasAccess(QueueACL.SUBMIT_APPLICATIONS, user));
    assertTrue(
        hasQueueACL(aclInfos, QueueACL.SUBMIT_APPLICATIONS, "root.c.c1.c11"));

    //Queue c111 has SA and AQ, both from parent
    assertTrue(c111.hasAccess(QueueACL.ADMINISTER_QUEUE, user));
    assertTrue(hasQueueACL(
        aclInfos,  QueueACL.ADMINISTER_QUEUE, "root.c.c1.c11.c111"));
    assertTrue(c111.hasAccess(QueueACL.SUBMIT_APPLICATIONS, user));
    assertTrue(hasQueueACL(
        aclInfos, QueueACL.SUBMIT_APPLICATIONS, "root.c.c1.c11.c111"));

    reset(c);
  }

  @Test
  public void testAbsoluteResourceWithChangeInClusterResource()
      throws Exception {
    // Setup queue configs
    setupSingleLevelQueuesWithAbsoluteResource(csConf);

    CSQueueStore queues = new CSQueueStore();
    CSQueue root = CapacitySchedulerQueueManager.parseQueue(csContext, csConf,
        null, CapacitySchedulerConfiguration.ROOT, queues, queues,
        TestUtils.spyHook);

    // Setup some nodes
    final int memoryPerNode = 10;
    int coresPerNode = 16;
    int numNodes = 2;

    Resource clusterResource = Resources.createResource(
        numNodes * (memoryPerNode * GB), numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    // Start testing
    LeafQueue a = (LeafQueue) queues.get(A);
    LeafQueue b = (LeafQueue) queues.get(B);

    assertEquals(a.getQueueResourceQuotas().getConfiguredMinResource(),
        QUEUE_A_RESOURCE);
    assertEquals(b.getQueueResourceQuotas().getConfiguredMinResource(),
        QUEUE_B_RESOURCE);
    assertEquals(a.getQueueResourceQuotas().getEffectiveMinResource(),
        QUEUE_A_RESOURCE);
    assertEquals(b.getQueueResourceQuotas().getEffectiveMinResource(),
        QUEUE_B_RESOURCE);

    numNodes = 1;
    clusterResource = Resources.createResource(numNodes * (memoryPerNode * GB),
        numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    Resource QUEUE_B_RESOURCE_HALF = Resource.newInstance(7 * 1024, 11);
    Resource QUEUE_A_RESOURCE_HALF = Resource.newInstance(3 * 1024, 5);
    assertEquals(a.getQueueResourceQuotas().getConfiguredMinResource(),
        QUEUE_A_RESOURCE);
    assertEquals(b.getQueueResourceQuotas().getConfiguredMinResource(),
        QUEUE_B_RESOURCE);
    assertEquals(a.getQueueResourceQuotas().getEffectiveMinResource(),
        QUEUE_A_RESOURCE_HALF);
    assertEquals(b.getQueueResourceQuotas().getEffectiveMinResource(),
        QUEUE_B_RESOURCE_HALF);

    coresPerNode = 40;
    clusterResource = Resources.createResource(numNodes * (memoryPerNode * GB),
        numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    Resource QUEUE_B_RESOURCE_70PERC = Resource.newInstance(7 * 1024, 27);
    Resource QUEUE_A_RESOURCE_30PERC = Resource.newInstance(3 * 1024, 12);
    assertEquals(a.getQueueResourceQuotas().getConfiguredMinResource(),
        QUEUE_A_RESOURCE);
    assertEquals(b.getQueueResourceQuotas().getConfiguredMinResource(),
        QUEUE_B_RESOURCE);
    assertEquals(a.getQueueResourceQuotas().getEffectiveMinResource(),
        QUEUE_A_RESOURCE_30PERC);
    assertEquals(b.getQueueResourceQuotas().getEffectiveMinResource(),
        QUEUE_B_RESOURCE_70PERC);
  }

  @Test
  public void testDeriveCapacityFromAbsoluteConfigurations() throws Exception {
    // Setup queue configs
    setupSingleLevelQueuesWithAbsoluteResource(csConf);

    CSQueueStore queues = new CSQueueStore();
    CSQueue root = CapacitySchedulerQueueManager.parseQueue(csContext, csConf,
            null, CapacitySchedulerConfiguration.ROOT, queues, queues,
            TestUtils.spyHook);

    // Setup some nodes
    int numNodes = 2;
    final long memoryPerNode = (QUEUE_A_RESOURCE.getMemorySize() +
            QUEUE_B_RESOURCE.getMemorySize()) / numNodes;
    int coresPerNode = (QUEUE_A_RESOURCE.getVirtualCores() +
            QUEUE_B_RESOURCE.getVirtualCores()) / numNodes;

    Resource clusterResource = Resources.createResource(
            numNodes * memoryPerNode, numNodes * coresPerNode);
    when(csContext.getNumClusterNodes()).thenReturn(numNodes);
    root.updateClusterResource(clusterResource,
            new ResourceLimits(clusterResource));

    // Start testing
    // Only MaximumSystemApplications is set in csConf
    LeafQueue a = (LeafQueue) queues.get(A);
    LeafQueue b = (LeafQueue) queues.get(B);

    float queueAScale = (float) QUEUE_A_RESOURCE.getMemorySize() /
            (float) clusterResource.getMemorySize();
    float queueBScale = (float) QUEUE_B_RESOURCE.getMemorySize() /
            (float) clusterResource.getMemorySize();

    assertEquals(queueAScale, a.getQueueCapacities().getCapacity(),
        DELTA);
    assertEquals(1f, a.getQueueCapacities().getMaximumCapacity(),
        DELTA);
    assertEquals(queueAScale, a.getQueueCapacities().getAbsoluteCapacity(),
        DELTA);
    assertEquals(1f,
        a.getQueueCapacities().getAbsoluteMaximumCapacity(), DELTA);
    assertEquals((int) (csConf.getMaximumSystemApplications() * queueAScale),
            a.getMaxApplications());
    assertEquals(a.getMaxApplications(), a.getMaxApplicationsPerUser());

    assertEquals(queueBScale,
        b.getQueueCapacities().getCapacity(), DELTA);
    assertEquals(1f,
        b.getQueueCapacities().getMaximumCapacity(), DELTA);
    assertEquals(queueBScale,
        b.getQueueCapacities().getAbsoluteCapacity(), DELTA);
    assertEquals(1f,
        b.getQueueCapacities().getAbsoluteMaximumCapacity(), DELTA);
    assertEquals((int) (csConf.getMaximumSystemApplications() * queueBScale),
            b.getMaxApplications());
    assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());

    // Set GlobalMaximumApplicationsPerQueue in csConf
    csConf.setGlobalMaximumApplicationsPerQueue(20000);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    assertEquals((int) (csConf.getGlobalMaximumApplicationsPerQueue() *
            queueAScale), a.getMaxApplications());
    assertEquals(a.getMaxApplications(), a.getMaxApplicationsPerUser());
    assertEquals((int) (csConf.getGlobalMaximumApplicationsPerQueue() *
            queueBScale), b.getMaxApplications());
    assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());

    // Set MaximumApplicationsPerQueue in csConf
    int queueAMaxApplications = 30000;
    int queueBMaxApplications = 30000;
    csConf.set("yarn.scheduler.capacity." + Q_A + ".maximum-applications",
            Integer.toString(queueAMaxApplications));
    csConf.set("yarn.scheduler.capacity." + Q_B + ".maximum-applications",
            Integer.toString(queueBMaxApplications));
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    assertEquals(queueAMaxApplications, a.getMaxApplications());
    assertEquals(a.getMaxApplications(), a.getMaxApplicationsPerUser());
    assertEquals(queueBMaxApplications, b.getMaxApplications());
    assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());

    // Extra cases for testing maxApplicationsPerUser
    int halfPercent = 50;
    int oneAndQuarterPercent = 125;
    a.getUsersManager().setUserLimit(halfPercent);
    b.getUsersManager().setUserLimit(oneAndQuarterPercent);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    assertEquals(a.getMaxApplications() * halfPercent / 100,
            a.getMaxApplicationsPerUser());
    // Q_B's limit per user shouldn't be greater
    // than the whole queue's application limit
    assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());

    float userLimitFactorQueueA = 0.9f;
    float userLimitFactorQueueB = 1.1f;
    a.getUsersManager().setUserLimit(halfPercent);
    a.getUsersManager().setUserLimitFactor(userLimitFactorQueueA);
    b.getUsersManager().setUserLimit(100);
    b.getUsersManager().setUserLimitFactor(userLimitFactorQueueB);
    root.updateClusterResource(clusterResource,
        new ResourceLimits(clusterResource));

    assertEquals((int) (a.getMaxApplications() * halfPercent *
            userLimitFactorQueueA / 100), a.getMaxApplicationsPerUser());
    // Q_B's limit per user shouldn't be greater
    // than the whole queue's application limit
    assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser());

  }

  @After
  public void tearDown() throws Exception {
  }

  private ResourceLimits anyResourceLimits() {
    return any(ResourceLimits.class);
  }
}