CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
    ResourceAllocationCommitter, MutableConfScheduler { public void markContainerForKillable( RMContainer killableContainer) { try { writeLock.lock(); if (LOG.isDebugEnabled()) { LOG.debug(SchedulerEventType.MARK_CONTAINER_FOR_KILLABLE + ": container" + killableContainer.toString()); } if (!isLazyPreemptionEnabled) { super.completedContainer(killableContainer, SchedulerUtils .createPreemptedContainerStatus(killableContainer.getContainerId(), SchedulerUtils.PREEMPTED_CONTAINER), RMContainerEventType.KILL); } else{ FiCaSchedulerNode node = (FiCaSchedulerNode) getSchedulerNode( killableContainer.getAllocatedNode()); FiCaSchedulerApp application = getCurrentAttemptForContainer( killableContainer.getContainerId()); node.markContainerToKillable(killableContainer.getContainerId()); if (null != application) { String leafQueueName = application.getCSLeafQueue().getQueueName(); getPreemptionManager().addKillableContainer( new KillableContainer(killableContainer, node.getPartition(), leafQueueName)); } } } finally { writeLock.unlock(); } } CapacityScheduler(); @Override void setConf(Configuration conf); @Override Configuration getConf(); @Override QueueMetrics getRootQueueMetrics(); CSQueue getRootQueue(); @Override CapacitySchedulerConfiguration getConfiguration(); @Override RMContainerTokenSecretManager getContainerTokenSecretManager(); @Override ResourceCalculator getResourceCalculator(); @VisibleForTesting void setResourceCalculator(ResourceCalculator rc); @Override int getNumClusterNodes(); @Override RMContext getRMContext(); @Override void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override void reinitialize(Configuration newConf, RMContext rmContext); @VisibleForTesting PlacementRule getUserGroupMappingPlacementRule(); PlacementRule getAppNameMappingPlacementRule(); @VisibleForTesting void updatePlacementRules(); @Override CSQueue getQueue(String queueName); @Override @Lock(Lock.NoLock.class) Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests,
      List<ContainerId> release, List<String> blacklistAdditions,
      List<String> blacklistRemovals, ContainerUpdates updateRequests); @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); @Lock(Lock.NoLock.class) List<FiCaSchedulerNode> getAllNodes(); @Override @Lock(Lock.NoLock.class) void recover(RMState state); @Override void killReservedContainer(RMContainer container); @Override void markContainerForPreemption(ApplicationAttemptId aid,
      RMContainer cont); @VisibleForTesting @Override void killContainer(RMContainer container); void markContainerForKillable(
      RMContainer killableContainer); @Override boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); boolean isSystemAppsLimitReached(); @Override void removeQueue(String queueName); @Override void addQueue(Queue queue); @Override void setEntitlement(String inQueue, QueueEntitlement entitlement); @Override String moveApplication(ApplicationId appId,
      String targetQueueName); @Override void preValidateMoveApplication(ApplicationId appId,
      String newQueue); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Resource getMaximumResourceCapability(String queueName); @Override Set<String> getPlanQueues(); @Override Priority checkAndGetApplicationPriority(
      Priority priorityRequestedByApp, UserGroupInformation user,
      String queueName, ApplicationId applicationId); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); @Override PreemptionManager getPreemptionManager(); @Override ResourceUsage getClusterResourceUsage(); @VisibleForTesting void submitResourceCommitRequest(Resource cluster,
      CSAssignment csAssignment); @Override boolean attemptAllocationOnNode(SchedulerApplicationAttempt appAttempt,
      SchedulingRequest schedulingRequest, SchedulerNode schedulerNode); @VisibleForTesting ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode> createResourceCommitRequest(CSAssignment csAssignment); @Override boolean tryCommit(Resource cluster, ResourceCommitRequest r,
      boolean updatePending); int getAsyncSchedulingPendingBacklogs(); @Override CapacitySchedulerQueueManager getCapacitySchedulerQueueManager(); boolean moveReservedContainer(RMContainer toBeMovedContainer,
      FiCaSchedulerNode targetNode); @Override long checkAndGetApplicationLifetime(String queueName,
      long lifetimeRequestedByApp); @Override long getMaximumApplicationLifetime(String queueName); @Override boolean isConfigurationMutable(); @Override MutableConfigurationProvider getMutableConfProvider(); @Override void resetSchedulerMetrics();  }
@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 = MockRM.launchAM(app0, rm1, nm1); am0.registerAppAttempt(); FiCaSchedulerApp schedulerAppAttempt = cs.getSchedulerApplications().get(app0.getApplicationId()) .getCurrentAppAttempt(); List<Container> allocatedContainers = am0.allocateAndWaitForContainers(3, CONTAINER_MEMORY, nm1); for (Container c : allocatedContainers) { cs.markContainerForKillable(schedulerAppAttempt.getRMContainer(c.getId())); } waitForAppPreemptionInfo(app0, Resource.newInstance(CONTAINER_MEMORY * 3, 3), 0, 3, Resource.newInstance(CONTAINER_MEMORY * 3, 3), false, 3); cs.markContainerForKillable(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 = MockRM.launchAM(app0, rm1, nm1); am1.registerAppAttempt(); schedulerAppAttempt = cs.getSchedulerApplications().get(app0.getApplicationId()) .getCurrentAppAttempt(); allocatedContainers = am1.allocateAndWaitForContainers(3, CONTAINER_MEMORY, nm1); for (Container c : allocatedContainers) { cs.markContainerForKillable(schedulerAppAttempt.getRMContainer(c.getId())); } waitForAppPreemptionInfo(app0, Resource.newInstance(CONTAINER_MEMORY * 7, 7), 1, 6, Resource.newInstance(CONTAINER_MEMORY * 3, 3), false, 3); rm1.stop(); }