FifoScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    Configurable { @Override public Allocation allocate(ApplicationAttemptId applicationAttemptId, List<ResourceRequest> ask, List<ContainerId> release, List<String> blacklistAdditions, List<String> blacklistRemovals, List<ContainerResourceChangeRequest> increaseRequests, List<ContainerResourceChangeRequest> decreaseRequests) { FiCaSchedulerApp application = getApplicationAttempt(applicationAttemptId); if (application == null) { LOG.error("Calling allocate on removed " + "or non existant application " + applicationAttemptId); return EMPTY_ALLOCATION; } SchedulerUtils.normalizeRequests(ask, resourceCalculator, clusterResource, minimumAllocation, getMaximumResourceCapability()); releaseContainers(release, application); synchronized (application) { if (application.isStopped()) { LOG.info("Calling allocate on a stopped " + "application " + applicationAttemptId); return EMPTY_ALLOCATION; } if (!ask.isEmpty()) { LOG.debug("allocate: pre-update" + " applicationId=" + applicationAttemptId + " application=" + application); application.showRequests(); application.updateResourceRequests(ask); LOG.debug("allocate: post-update" + " applicationId=" + applicationAttemptId + " application=" + application); application.showRequests(); LOG.debug("allocate:" + " applicationId=" + applicationAttemptId + " #ask=" + ask.size()); } if (application.isWaitingForAMContainer()) { application.updateAMBlacklist( blacklistAdditions, blacklistRemovals); } else { application.updateBlacklist(blacklistAdditions, blacklistRemovals); } Resource headroom = application.getHeadroom(); application.setApplicationHeadroomForMetrics(headroom); return new Allocation(application.pullNewlyAllocatedContainers(), headroom, null, null, null, application.pullUpdatedNMTokens()); } } FifoScheduler(); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void setConf(Configuration conf); @Override synchronized Configuration getConf(); @Override int getNumClusterNodes(); @Override synchronized void setRMContext(RMContext rmContext); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @Override Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release,
      List<String> blacklistAdditions, List<String> blacklistRemovals,
      List<ContainerResourceChangeRequest> increaseRequests,
      List<ContainerResourceChangeRequest> decreaseRequests); @VisibleForTesting synchronized void addApplication(ApplicationId applicationId,
      String queue, String user, boolean isAppRecovering); @VisibleForTesting synchronized void addApplicationAttempt(ApplicationAttemptId appAttemptId,
          boolean transferStateFromPreviousAttempt,
          boolean isAttemptRecovering); @Override void handle(SchedulerEvent event); @Override QueueInfo getQueueInfo(String queueName,
      boolean includeChildQueues, boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override ResourceCalculator getResourceCalculator(); @Override void recover(RMState state); @Override RMContainer getRMContainer(ContainerId containerId); @Override QueueMetrics getRootQueueMetrics(); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override synchronized List<ApplicationAttemptId> getAppsInQueue(String queueName); Resource getUsedResource();  }
@Test(timeout = 60000) public void testAllocateContainerOnNodeWithoutOffSwitchSpecified() throws Exception { Logger rootLogger = LogManager.getRootLogger(); rootLogger.setLevel(Level.DEBUG); MockRM rm = new MockRM(conf); rm.start(); MockNM nm1 = rm.registerNode("127.0.0.1:1234", 6 * GB); RMApp app1 = rm.submitApp(2048); nm1.nodeHeartbeat(true); RMAppAttempt attempt1 = app1.getCurrentAppAttempt(); MockAM am1 = rm.sendAMLaunched(attempt1.getAppAttemptId()); am1.registerAppAttempt(); List<ResourceRequest> requests = new ArrayList<ResourceRequest>(); requests.add(am1.createResourceReq("127.0.0.1", 1 * GB, 1, 1)); requests.add(am1.createResourceReq("/default-rack", 1 * GB, 1, 1)); am1.allocate(requests, null); try { nm1.nodeHeartbeat(true); } catch (NullPointerException e) { Assert.fail("NPE when allocating container on node but " + "forget to set off-switch request should be handled"); } rm.stop(); }