@Test(timeout = 120000) public void testPreemptionInfo() throws Exception { Configuration conf = new Configuration(); conf.setInt(YarnConfiguration.RM_AM_MAX_ATTEMPTS, 3); conf.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); int CONTAINER_MEMORY = 1024; MockRM rm1 = new MockRM(conf); rm1.start(); CapacityScheduler cs = (CapacityScheduler) rm1.getResourceScheduler(); MockNM nm1 = new MockNM("127.0.0.1:1234", 15120, rm1.getResourceTrackerService()); nm1.registerNode(); RMApp app0 = rm1.submitApp(CONTAINER_MEMORY); MockAM am0 = launchAM(app0, rm1, nm1); FiCaSchedulerApp schedulerAppAttempt = cs.getSchedulerApplications().get(app0.getApplicationId()) .getCurrentAppAttempt(); List<Container> allocatedContainers = am0.allocateAndWaitForContainers(3, CONTAINER_MEMORY, nm1); for (Container c : allocatedContainers) { cs.killContainer(schedulerAppAttempt.getRMContainer(c.getId())); } waitForAppPreemptionInfo(app0, Resource.newInstance(CONTAINER_MEMORY * 3, 3), 0, 3, Resource.newInstance(CONTAINER_MEMORY * 3, 3), false, 3); cs.killContainer(schedulerAppAttempt.getRMContainer(app0 .getCurrentAppAttempt().getMasterContainer().getId())); waitForNewAttemptCreated(app0, am0.getApplicationAttemptId()); waitForAppPreemptionInfo(app0, Resource.newInstance(CONTAINER_MEMORY * 4, 4), 1, 3, Resource.newInstance(0, 0), false, 0); MockAM am1 = launchAM(app0, rm1, nm1); schedulerAppAttempt = cs.getSchedulerApplications().get(app0.getApplicationId()) .getCurrentAppAttempt(); allocatedContainers = am1.allocateAndWaitForContainers(3, CONTAINER_MEMORY, nm1); for (Container c : allocatedContainers) { cs.killContainer(schedulerAppAttempt.getRMContainer(c.getId())); } waitForAppPreemptionInfo(app0, Resource.newInstance(CONTAINER_MEMORY * 7, 7), 1, 6, Resource.newInstance(CONTAINER_MEMORY * 3, 3), false, 3); rm1.stop(); }
@Override public void killContainer(RMContainer cont) { if (LOG.isDebugEnabled()) { LOG.debug("KILL_CONTAINER: container" + cont.toString()); } recoverResourceRequestForContainer(cont); completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus( cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL); }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { @Override public void killContainer(RMContainer cont) { if (LOG.isDebugEnabled()) { LOG.debug("KILL_CONTAINER: container" + cont.toString()); } recoverResourceRequestForContainer(cont); completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus( cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL); } }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { @Override public void killContainer(RMContainer cont) { if (LOG.isDebugEnabled()) { LOG.debug("KILL_CONTAINER: container" + cont.toString()); } recoverResourceRequestForContainer(cont); completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus( cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL); } CapacityScheduler(); }
CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { @Override public void killContainer(RMContainer cont) { if (LOG.isDebugEnabled()) { LOG.debug("KILL_CONTAINER: container" + cont.toString()); } recoverResourceRequestForContainer(cont); completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus( cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL); } 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 { @Override public void killContainer(RMContainer cont) { if (LOG.isDebugEnabled()) { LOG.debug("KILL_CONTAINER: container" + cont.toString()); } recoverResourceRequestForContainer(cont); completedContainer(cont, SchedulerUtils.createPreemptedContainerStatus( cont.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL); } 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; }