LeafQueue extends AbstractCSQueue { @Override public void submitApplicationAttempt(FiCaSchedulerApp application, String userName) { synchronized (this) { User user = getUser(userName); addApplicationAttempt(application, user); } if (application.isPending()) { metrics.submitAppAttempt(userName); } getParent().submitApplicationAttempt(application, userName); } LeafQueue(CapacitySchedulerContext cs, 
      String queueName, CSQueue parent, CSQueue old); @Override String getQueuePath(); @Private float getMinimumAllocationFactor(); @Private float getMaxAMResourcePerQueuePercent(); int getMaxApplications(); synchronized int getMaxApplicationsPerUser(); @Override ActiveUsersManager getActiveUsersManager(); @Override List<CSQueue> getChildQueues(); @Override synchronized int getNumApplications(); synchronized int getNumPendingApplications(); synchronized int getNumActiveApplications(); @Private synchronized int getNumApplications(String user); @Private synchronized int getNumPendingApplications(String user); @Private synchronized int getNumActiveApplications(String user); synchronized int getNumContainers(); @Override synchronized QueueState getState(); @Private synchronized int getUserLimit(); @Private synchronized float getUserLimitFactor(); @Override synchronized QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive); @Override synchronized List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); @Private int getNodeLocalityDelay(); String toString(); @VisibleForTesting synchronized void setNodeLabelManager(RMNodeLabelsManager mgr); @VisibleForTesting synchronized User getUser(String userName); synchronized ArrayList<UserInfo> getUsers(); @Override synchronized void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName); @Override void submitApplication(ApplicationId applicationId, String userName,
      String queue); synchronized Resource getAMResourceLimit(); synchronized Resource getUserAMResourceLimit(); @Override void finishApplication(ApplicationId application, String user); @Override void finishApplicationAttempt(FiCaSchedulerApp application, String queue); synchronized void removeApplicationAttempt(
      FiCaSchedulerApp application, User user); @Override CSAssignment assignContainers(Resource clusterResource,
      FiCaSchedulerNode node, boolean needToUnreserve); @Override void completedContainer(Resource clusterResource, 
      FiCaSchedulerApp application, FiCaSchedulerNode node, RMContainer rmContainer, 
      ContainerStatus containerStatus, RMContainerEventType event, CSQueue childQueue,
      boolean sortQueues); @Override synchronized void updateClusterResource(Resource clusterResource); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt attempt, RMContainer rmContainer); Set<FiCaSchedulerApp> getApplications(); synchronized Resource getTotalResourcePending(); @Override synchronized void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override void attachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); @Override void detachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); void setCapacity(float capacity); void setAbsoluteCapacity(float absoluteCapacity); void setMaxApplications(int maxApplications);  }
@Test public void testUserQueueAcl() throws Exception { LeafQueue d = stubLeafQueue((LeafQueue) queues.get(D)); final String user_d = "user_d"; final ApplicationAttemptId appAttemptId_0 = TestUtils .getMockApplicationAttemptId(0, 1); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_d, d, null, spyRMContext); d.submitApplicationAttempt(app_0, user_d); final ApplicationAttemptId appAttemptId_1 = TestUtils .getMockApplicationAttemptId(0, 2); FiCaSchedulerApp app_1 = new FiCaSchedulerApp(appAttemptId_1, user_d, d, null, spyRMContext); d.submitApplicationAttempt(app_1, user_d); }
@Test public void testAppAttemptMetrics() throws Exception { LeafQueue a = stubLeafQueue((LeafQueue) queues.get(B)); final String user_0 = "user_0"; final ApplicationAttemptId appAttemptId_0 = TestUtils .getMockApplicationAttemptId(0, 1); AppAddedSchedulerEvent addAppEvent = new AppAddedSchedulerEvent(appAttemptId_0.getApplicationId(), a.getQueueName(), user_0); cs.handle(addAppEvent); AppAttemptAddedSchedulerEvent addAttemptEvent = new AppAttemptAddedSchedulerEvent(appAttemptId_0, false); cs.handle(addAttemptEvent); AppAttemptRemovedSchedulerEvent event = new AppAttemptRemovedSchedulerEvent( appAttemptId_0, RMAppAttemptState.FAILED, false); cs.handle(event); assertEquals(0, a.getMetrics().getAppsPending()); assertEquals(0, a.getMetrics().getAppsFailed()); final ApplicationAttemptId appAttemptId_1 = TestUtils .getMockApplicationAttemptId(0, 2); FiCaSchedulerApp app_1 = new FiCaSchedulerApp(appAttemptId_1, user_0, a, null, spyRMContext); a.submitApplicationAttempt(app_1, user_0); assertEquals(1, a.getMetrics().getAppsSubmitted()); assertEquals(1, a.getMetrics().getAppsPending()); event = new AppAttemptRemovedSchedulerEvent(appAttemptId_0, RMAppAttemptState.FINISHED, false); cs.handle(event); AppRemovedSchedulerEvent rEvent = new AppRemovedSchedulerEvent( appAttemptId_0.getApplicationId(), RMAppState.FINISHED); cs.handle(rEvent); assertEquals(1, a.getMetrics().getAppsSubmitted()); assertEquals(0, a.getMetrics().getAppsPending()); assertEquals(0, a.getMetrics().getAppsFailed()); assertEquals(1, a.getMetrics().getAppsCompleted()); QueueMetrics userMetrics = a.getMetrics().getUserMetrics(user_0); assertEquals(1, userMetrics.getAppsSubmitted()); }