FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { public RMContext getRMContext() { return rmContext; } FairScheduler(); FSContext getContext(); RMContext getRMContext(); boolean isAtLeastReservationThreshold(
      ResourceCalculator resourceCalculator, Resource resource); FairSchedulerConfiguration getConf(); int getNumNodesInRack(String rackName); QueueManager getQueueManager(); @VisibleForTesting @Override void update(); RMContainerTokenSecretManager getContainerTokenSecretManager(); boolean isSizeBasedWeight(); Resource getIncrementResourceCapability(); double getNodeLocalityThreshold(); double getRackLocalityThreshold(); @Deprecated long getNodeLocalityDelayMs(); @Deprecated long getRackLocalityDelayMs(); @Deprecated boolean isContinuousSchedulingEnabled(); @Deprecated int getContinuousSchedulingSleepMs(); @Override Resource getNormalizedResource(Resource requestedResource,
      Resource maxResourceCapability); @Override Resource getMaximumResourceCapability(String queueName); @VisibleForTesting @Override void killContainer(RMContainer container); @Override Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests,
      List<ContainerId> release, List<String> blacklistAdditions,
      List<String> blacklistRemovals, ContainerUpdates updateRequests); FSAppAttempt getSchedulerApp(ApplicationAttemptId appAttemptId); @Override ResourceCalculator getResourceCalculator(); @Override QueueMetrics getRootQueueMetrics(); @Override void handle(SchedulerEvent event); @Override void recover(RMState state); void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @SuppressWarnings("deprecation") @Override void serviceStop(); @Override void reinitialize(Configuration conf, RMContext rmContext); @Override QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override int getNumClusterNodes(); @Override boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); AllocationConfiguration getAllocationConfiguration(); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override String moveApplication(ApplicationId appId,
      String queueName); @Override void preValidateMoveApplication(ApplicationId appId, String newQueue); @Override void updateNodeResource(RMNode nm,
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); @Override void setEntitlement(String queueName,
      QueueEntitlement entitlement); @Override void removeQueue(String queueName); float getReservableNodesRatio(); @Override long checkAndGetApplicationLifetime(String queueName, long lifetime); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); boolean isNoTerminalRuleCheck(); static final Resource CONTAINER_RESERVED; }
@Test(timeout = 120000) public void testRefreshQueuesWhenRMHA() throws Exception { conf.setBoolean(YarnConfiguration.AUTO_FAILOVER_ENABLED, false); conf.setBoolean(YarnConfiguration.RECOVERY_ENABLED, true); conf.setBoolean(FairSchedulerConfiguration.ALLOW_UNDECLARED_POOLS, false); conf.setBoolean(FairSchedulerConfiguration.USER_AS_DEFAULT_QUEUE, false); conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE); HAServiceProtocol.StateChangeRequestInfo requestInfo = new HAServiceProtocol.StateChangeRequestInfo( HAServiceProtocol.RequestSource.REQUEST_BY_USER); AllocationFileWriter.create().writeToFile(ALLOC_FILE); MockRM rm1 = new MockRM(conf, null); rm1.init(conf); rm1.start(); rm1.getAdminService().transitionToStandby(requestInfo); AllocationFileWriter.create() .addQueue(new AllocationFileQueue.Builder("test_queue") .maxRunningApps(3).build()) .writeToFile(ALLOC_FILE); conf.set(YarnConfiguration.RM_STORE, MemoryRMStateStore.class.getName()); MockRM rm2 = new MockRM(conf); MemoryRMStateStore memStore = (MemoryRMStateStore) rm2.getRMStateStore(); rm2.start(); MockNM nm = new MockNM("127.0.0.1:1234", 15120, rm2.getResourceTrackerService()); nm.registerNode(); rm2.getAdminService().transitionToActive(requestInfo); MockRMAppSubmissionData data = MockRMAppSubmissionData.Builder.createWithMemory(200, rm2) .withAppName("test_app") .withUser("user") .withAcls(null) .withQueue("test_queue") .withUnmanagedAM(false) .build(); RMApp app = MockRMAppSubmitter.submit(rm2, data); RMAppAttempt attempt0 = app.getCurrentAppAttempt(); nm.nodeHeartbeat(true); MockAM am0 = rm2.sendAMLaunched(attempt0.getAppAttemptId()); am0.registerAppAttempt(); assertEquals("root.test_queue", app.getQueue()); ((RMContextImpl)rm1.getRMContext()).setStateStore(memStore); rm1.getAdminService().transitionToActive(requestInfo); rm1.drainEvents(); assertEquals(1, rm1.getRMContext().getRMApps().size()); RMApp recoveredApp = rm1.getRMContext().getRMApps().values().iterator().next(); assertEquals("root.test_queue", recoveredApp.getQueue()); rm1.stop(); rm2.stop(); }