@Test public void testUpdateDemand() { conf.set(FairSchedulerConfiguration.ASSIGN_MULTIPLE, "false"); resourceManager = new MockRM(conf); resourceManager.start(); scheduler = (FairScheduler) resourceManager.getResourceScheduler(); scheduler.allocConf = mock(AllocationConfiguration.class); String queueName = "root.queue1"; when(scheduler.allocConf.getMaxResources(queueName)).thenReturn(maxResource); when(scheduler.allocConf.getMinResources(queueName)).thenReturn(Resources.none()); FSLeafQueue schedulable = new FSLeafQueue(queueName, scheduler, null); FSAppAttempt app = mock(FSAppAttempt.class); Mockito.when(app.getDemand()).thenReturn(maxResource); schedulable.addAppSchedulable(app); schedulable.addAppSchedulable(app); schedulable.updateDemand(); assertTrue("Demand is greater than max allowed ", Resources.equals(schedulable.getDemand(), maxResource)); }
@Override public void updateDemand() { Resource maxRes = scheduler.getAllocationConfiguration() .getMaxResources(getName()); demand = Resources.createResource(0); readLock.lock(); try { for (FSAppAttempt sched : runnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } for (FSAppAttempt sched : nonRunnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } } finally { readLock.unlock(); } if (LOG.isDebugEnabled()) { LOG.debug("The updated demand for " + getName() + " is " + demand + "; the max is " + maxRes); } }
FSLeafQueue extends FSQueue { @Override public void updateDemand() { Resource maxRes = scheduler.getAllocationConfiguration() .getMaxResources(getName()); demand = Resources.createResource(0); readLock.lock(); try { for (FSAppAttempt sched : runnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } for (FSAppAttempt sched : nonRunnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } } finally { readLock.unlock(); } if (LOG.isDebugEnabled()) { LOG.debug("The updated demand for " + getName() + " is " + demand + "; the max is " + maxRes); } } }
FSLeafQueue extends FSQueue { @Override public void updateDemand() { Resource maxRes = scheduler.getAllocationConfiguration() .getMaxResources(getName()); demand = Resources.createResource(0); readLock.lock(); try { for (FSAppAttempt sched : runnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } for (FSAppAttempt sched : nonRunnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } } finally { readLock.unlock(); } if (LOG.isDebugEnabled()) { LOG.debug("The updated demand for " + getName() + " is " + demand + "; the max is " + maxRes); } } FSLeafQueue(String name, FairScheduler scheduler,
      FSParentQueue parent); }
FSLeafQueue extends FSQueue { @Override public void updateDemand() { Resource maxRes = scheduler.getAllocationConfiguration() .getMaxResources(getName()); demand = Resources.createResource(0); readLock.lock(); try { for (FSAppAttempt sched : runnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } for (FSAppAttempt sched : nonRunnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } } finally { readLock.unlock(); } if (LOG.isDebugEnabled()) { LOG.debug("The updated demand for " + getName() + " is " + demand + "; the max is " + maxRes); } } FSLeafQueue(String name, FairScheduler scheduler,
      FSParentQueue parent); void addApp(FSAppAttempt app, boolean runnable); boolean removeApp(FSAppAttempt app); boolean removeNonRunnableApp(FSAppAttempt app); boolean isRunnableApp(FSAppAttempt attempt); boolean isNonRunnableApp(FSAppAttempt attempt); void resetPreemptedResources(); void clearPreemptedResources(); List<FSAppAttempt> getCopyOfNonRunnableAppSchedulables(); @Override void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override void setPolicy(SchedulingPolicy policy); @Override void recomputeShares(); @Override Resource getDemand(); @Override Resource getResourceUsage(); Resource getAmResourceUsage(); @Override void updateDemand(); @Override Resource assignContainer(FSSchedulerNode node); @Override RMContainer preemptContainer(); @Override List<FSQueue> getChildQueues(); @Override List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); long getLastTimeAtMinShare(); long getLastTimeAtFairShareThreshold(); @Override int getNumRunnableApps(); int getNumNonRunnableApps(); int getNumPendingApps(); int getNumActiveApps(); @Override ActiveUsersManager getActiveUsersManager(); boolean canRunAppAM(Resource amResource); void addAMResourceUsage(Resource amResource); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt schedulerAttempt, RMContainer rmContainer); void updateStarvationStats(); }
FSLeafQueue extends FSQueue { @Override public void updateDemand() { Resource maxRes = scheduler.getAllocationConfiguration() .getMaxResources(getName()); demand = Resources.createResource(0); readLock.lock(); try { for (FSAppAttempt sched : runnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } for (FSAppAttempt sched : nonRunnableApps) { if (Resources.equals(demand, maxRes)) { break; } updateDemandForApp(sched, maxRes); } } finally { readLock.unlock(); } if (LOG.isDebugEnabled()) { LOG.debug("The updated demand for " + getName() + " is " + demand + "; the max is " + maxRes); } } FSLeafQueue(String name, FairScheduler scheduler,
      FSParentQueue parent); void addApp(FSAppAttempt app, boolean runnable); boolean removeApp(FSAppAttempt app); boolean removeNonRunnableApp(FSAppAttempt app); boolean isRunnableApp(FSAppAttempt attempt); boolean isNonRunnableApp(FSAppAttempt attempt); void resetPreemptedResources(); void clearPreemptedResources(); List<FSAppAttempt> getCopyOfNonRunnableAppSchedulables(); @Override void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override void setPolicy(SchedulingPolicy policy); @Override void recomputeShares(); @Override Resource getDemand(); @Override Resource getResourceUsage(); Resource getAmResourceUsage(); @Override void updateDemand(); @Override Resource assignContainer(FSSchedulerNode node); @Override RMContainer preemptContainer(); @Override List<FSQueue> getChildQueues(); @Override List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); long getLastTimeAtMinShare(); long getLastTimeAtFairShareThreshold(); @Override int getNumRunnableApps(); int getNumNonRunnableApps(); int getNumPendingApps(); int getNumActiveApps(); @Override ActiveUsersManager getActiveUsersManager(); boolean canRunAppAM(Resource amResource); void addAMResourceUsage(Resource amResource); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt schedulerAttempt, RMContainer rmContainer); void updateStarvationStats();  }