@Test(timeout = 30000) public void testAppReservationWithDominantResourceCalculator() throws Exception { CapacitySchedulerConfiguration csconf = new CapacitySchedulerConfiguration(); csconf.setResourceComparator(DominantResourceCalculator.class); YarnConfiguration conf = new YarnConfiguration(csconf); conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 10 * GB, 1, 1); MockNM nm2 = rm.registerNode("127.0.0.1:1235", 10 * GB, 4, 4); rm.registerNode("127.0.0.1:1236", 10 * GB, 4, 4); RMApp app1 = rm.submitApp(1024); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); SchedulerNodeReport report_nm1 = rm.getResourceScheduler().getNodeReport(nm1.getNodeId()); Assert.assertEquals(1 * GB, report_nm1.getUsedResource().getMemory()); Assert.assertEquals(9 * GB, report_nm1.getAvailableResource().getMemory()); am1.addRequests(new String[] { "127.0.0.1", "127.0.0.2" }, 1 * GB, 1, 1); am1.schedule(); nm1.nodeHeartbeat(true); Thread.sleep(1000); AllocateResponse allocResponse = am1.schedule(); ApplicationResourceUsageReport report = rm.getResourceScheduler().getAppResourceUsageReport( attempt1.getAppAttemptId()); Assert.assertEquals(0, allocResponse.getAllocatedContainers().size()); Assert.assertEquals(0, report.getNumReservedContainers()); nm2.nodeHeartbeat(true); while (allocResponse.getAllocatedContainers().size() == 0) { Thread.sleep(100); allocResponse = am1.schedule(); } report = rm.getResourceScheduler().getAppResourceUsageReport( attempt1.getAppAttemptId()); Assert.assertEquals(1, allocResponse.getAllocatedContainers().size()); Assert.assertEquals(0, report.getNumReservedContainers()); rm.stop(); }
static void schedule(CapacityScheduler cs) { int current = 0; Collection<FiCaSchedulerNode> nodes = cs.getAllNodes().values(); int start = random.nextInt(nodes.size()); for (FiCaSchedulerNode node : nodes) { if (current++ >= start) { cs.allocateContainersToNode(node); } } for (FiCaSchedulerNode node : nodes) { cs.allocateContainersToNode(node); } try { Thread.sleep(cs.getAsyncScheduleInterval()); } catch (InterruptedException e) {} }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { static void schedule(CapacityScheduler cs) { int current = 0; Collection<FiCaSchedulerNode> nodes = cs.getAllNodes().values(); int start = random.nextInt(nodes.size()); for (FiCaSchedulerNode node : nodes) { if (current++ >= start) { cs.allocateContainersToNode(node); } } for (FiCaSchedulerNode node : nodes) { cs.allocateContainersToNode(node); } try { Thread.sleep(cs.getAsyncScheduleInterval()); } catch (InterruptedException e) {} } }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { static void schedule(CapacityScheduler cs) { int current = 0; Collection<FiCaSchedulerNode> nodes = cs.getAllNodes().values(); int start = random.nextInt(nodes.size()); for (FiCaSchedulerNode node : nodes) { if (current++ >= start) { cs.allocateContainersToNode(node); } } for (FiCaSchedulerNode node : nodes) { cs.allocateContainersToNode(node); } try { Thread.sleep(cs.getAsyncScheduleInterval()); } catch (InterruptedException e) {} } CapacityScheduler(); }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { static void schedule(CapacityScheduler cs) { int current = 0; Collection<FiCaSchedulerNode> nodes = cs.getAllNodes().values(); int start = random.nextInt(nodes.size()); for (FiCaSchedulerNode node : nodes) { if (current++ >= start) { cs.allocateContainersToNode(node); } } for (FiCaSchedulerNode node : nodes) { cs.allocateContainersToNode(node); } try { Thread.sleep(cs.getAsyncScheduleInterval()); } catch (InterruptedException e) {} } CapacityScheduler(); @Override void setConf(Configuration conf); @Override Configuration getConf(); @VisibleForTesting synchronized String getMappedQueueForTest(String user); @Override QueueMetrics getRootQueueMetrics(); CSQueue getRootQueue(); @Override CapacitySchedulerConfiguration getConfiguration(); @Override synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); @Override Comparator<FiCaSchedulerApp> getApplicationComparator(); @Override ResourceCalculator getResourceCalculator(); @Override Comparator<CSQueue> getQueueComparator(); @Override int getNumClusterNodes(); @Override synchronized RMContext getRMContext(); @Override synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @VisibleForTesting static void setQueueAcls(YarnAuthorizationProvider authorizer,
      Map<String, CSQueue> queues); CSQueue getQueue(String queueName); @Override @Lock(Lock.NoLock.class) Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release, 
      List<String> blacklistAdditions, List<String> blacklistRemovals); @Override @Lock(Lock.NoLock.class) QueueInfo getQueueInfo(String queueName, 
      boolean includeChildQueues, boolean recursive); @Override @Lock(Lock.NoLock.class) List<QueueUserACLInfo> getQueueUserAclInfo(); @Override void handle(SchedulerEvent event); @Lock(Lock.NoLock.class) @VisibleForTesting @Override FiCaSchedulerApp getApplicationAttempt(
      ApplicationAttemptId applicationAttemptId); @Lock(Lock.NoLock.class) FiCaSchedulerNode getNode(NodeId nodeId); @Override @Lock(Lock.NoLock.class) void recover(RMState state); @Override void dropContainerReservation(RMContainer container); @Override void preemptContainer(ApplicationAttemptId aid, RMContainer cont); @Override void killContainer(RMContainer cont); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized void removeQueue(String queueName); @Override synchronized void addQueue(Queue queue); @Override synchronized void setEntitlement(String inQueue,
      QueueEntitlement entitlement); @Override synchronized String moveApplication(ApplicationId appId,
      String targetQueueName); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Resource getMaximumResourceCapability(String queueName); @Override Set<String> getPlanQueues(); }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { static void schedule(CapacityScheduler cs) { int current = 0; Collection<FiCaSchedulerNode> nodes = cs.getAllNodes().values(); int start = random.nextInt(nodes.size()); for (FiCaSchedulerNode node : nodes) { if (current++ >= start) { cs.allocateContainersToNode(node); } } for (FiCaSchedulerNode node : nodes) { cs.allocateContainersToNode(node); } try { Thread.sleep(cs.getAsyncScheduleInterval()); } catch (InterruptedException e) {} } CapacityScheduler(); @Override void setConf(Configuration conf); @Override Configuration getConf(); @VisibleForTesting synchronized String getMappedQueueForTest(String user); @Override QueueMetrics getRootQueueMetrics(); CSQueue getRootQueue(); @Override CapacitySchedulerConfiguration getConfiguration(); @Override synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); @Override Comparator<FiCaSchedulerApp> getApplicationComparator(); @Override ResourceCalculator getResourceCalculator(); @Override Comparator<CSQueue> getQueueComparator(); @Override int getNumClusterNodes(); @Override synchronized RMContext getRMContext(); @Override synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @VisibleForTesting static void setQueueAcls(YarnAuthorizationProvider authorizer,
      Map<String, CSQueue> queues); CSQueue getQueue(String queueName); @Override @Lock(Lock.NoLock.class) Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release, 
      List<String> blacklistAdditions, List<String> blacklistRemovals); @Override @Lock(Lock.NoLock.class) QueueInfo getQueueInfo(String queueName, 
      boolean includeChildQueues, boolean recursive); @Override @Lock(Lock.NoLock.class) List<QueueUserACLInfo> getQueueUserAclInfo(); @Override void handle(SchedulerEvent event); @Lock(Lock.NoLock.class) @VisibleForTesting @Override FiCaSchedulerApp getApplicationAttempt(
      ApplicationAttemptId applicationAttemptId); @Lock(Lock.NoLock.class) FiCaSchedulerNode getNode(NodeId nodeId); @Override @Lock(Lock.NoLock.class) void recover(RMState state); @Override void dropContainerReservation(RMContainer container); @Override void preemptContainer(ApplicationAttemptId aid, RMContainer cont); @Override void killContainer(RMContainer cont); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized void removeQueue(String queueName); @Override synchronized void addQueue(Queue queue); @Override synchronized void setEntitlement(String inQueue,
      QueueEntitlement entitlement); @Override synchronized String moveApplication(ApplicationId appId,
      String targetQueueName); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Resource getMaximumResourceCapability(String queueName); @Override Set<String> getPlanQueues(); @Private
  static final String ROOT_QUEUE; }