FSAppAttempt extends SchedulerApplicationAttempt implements Schedulable { private Resource assignContainer( FSSchedulerNode node, PendingAsk pendingAsk, NodeType type, boolean reserved, SchedulerRequestKey schedulerKey) { Resource capability = pendingAsk.getPerAllocationResource(); Resource available = node.getUnallocatedResource(); Container reservedContainer = null; if (reserved) { reservedContainer = node.getReservedContainer().getContainer(); } if (Resources.fitsIn(capability, available)) { RMContainer allocatedContainer = allocate(type, node, schedulerKey, pendingAsk, reservedContainer); if (allocatedContainer == null) { if (reserved) { unreserve(schedulerKey, node); } LOG.debug("Resource ask {} fits in available node resources {}," + " but no container was allocated", capability, available); return Resources.none(); } if (reserved) { unreserve(schedulerKey, node); } node.allocateContainer(allocatedContainer); if (!isAmRunning() && !getUnmanagedAM()) { setAMResource(capability); getQueue().addAMResourceUsage(capability); setAmRunning(true); } return capability; } LOG.debug("Resource request: {} exceeds the available" + " resources of the node.", capability); if (isReservable(capability) && !node.isPreemptedForApp(this) && reserve(pendingAsk.getPerAllocationResource(), node, reservedContainer, type, schedulerKey)) { updateAMDiagnosticMsg(capability, " exceeds the available resources of " + "the node and the request is reserved)"); LOG.debug("{}'s resource request is reserved.", getName()); return FairScheduler.CONTAINER_RESERVED; } else { updateAMDiagnosticMsg(capability, " exceeds the available resources of " + "the node and the request cannot be reserved)"); if (LOG.isDebugEnabled()) { LOG.debug("Couldn't create reservation for app: " + getName() + ", at priority " + schedulerKey.getPriority()); } return Resources.none(); } } FSAppAttempt(FairScheduler scheduler,
      ApplicationAttemptId applicationAttemptId, String user, FSLeafQueue queue,
      ActiveUsersManager activeUsersManager, RMContext rmContext); QueueMetrics getMetrics(); @Override Resource getHeadroom(); RMContainer allocate(NodeType type, FSSchedulerNode node,
      SchedulerRequestKey schedulerKey, PendingAsk pendingAsk,
      Container reservedContainer); @Override FSLeafQueue getQueue(); @Override synchronized void recoverContainer(SchedulerNode node,
      RMContainer rmContainer); void unreserve(SchedulerRequestKey schedulerKey,
      FSSchedulerNode node); @Override String getName(); @Override Resource getDemand(); @Override long getStartTime(); @Override Resource getMinShare(); @Override Resource getMaxShare(); @Override Resource getResourceUsage(); @Override float getWeight(); @Override Priority getPriority(); @Override Resource getFairShare(); @Override void setFairShare(Resource fairShare); @Override void updateDemand(); @Override Resource assignContainer(FSSchedulerNode node); @Override int hashCode(); @Override boolean equals(Object o); @Override String toString(); @Override boolean isPreemptable(); @VisibleForTesting void setEnableAMPreemption(boolean enableAMPreemption);  }
@Test public void testNoNextPendingAsk() { FSLeafQueue queue = Mockito.mock(FSLeafQueue.class); ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1); RMContext rmContext = Mockito.mock(RMContext.class); ConcurrentMap<ApplicationId, RMApp> rmApps = new ConcurrentHashMap<>(); RMApp rmApp = Mockito.mock(RMApp.class); rmApps.put(applicationAttemptId.getApplicationId(), rmApp); ApplicationSubmissionContext appContext = Mockito.mock(ApplicationSubmissionContext.class); Mockito.when(appContext.getUnmanagedAM()).thenReturn(false); Mockito.when(appContext.getLogAggregationContext()) .thenReturn(Mockito.mock(LogAggregationContext.class)); Mockito.when(rmApp.getApplicationSchedulingEnvs()) .thenReturn(new HashMap<>()); Mockito.when(rmApp.getApplicationSubmissionContext()) .thenReturn(appContext); Mockito.when(rmContext.getRMApps()).thenReturn(rmApps); Mockito.when(rmContext.getYarnConfiguration()).thenReturn(conf); FSAppAttempt schedulerApp = new FSAppAttempt(scheduler, applicationAttemptId, "user1", queue, null, rmContext); schedulerApp.setAmRunning(false); FSSchedulerNode schedulerNode = Mockito.mock(FSSchedulerNode.class); Resource resource = schedulerApp.assignContainer(schedulerNode); assertEquals(Resources.none(), resource); }