AbstractYarnScheduler extends AbstractService implements ResourceScheduler { @Override public Resource getClusterResource() { return nodeTracker.getClusterCapacity(); } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); @VisibleForTesting ClusterNodeTracker<N> getNodeTracker(); @VisibleForTesting SchedulingMonitorManager getSchedulingMonitorManager(); List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); List<N> getBlacklistedNodes(final SchedulerApplicationAttempt app); List<N> getNodes(final NodeFilter filter); boolean shouldContainersBeAutoUpdated(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); @Override Resource getMaximumResourceCapability(String queueName); SchedulerHealth getSchedulerHealth(); long getLastNodeUpdateTime(); T getApplicationAttempt(ApplicationAttemptId applicationAttemptId); @Override SchedulerAppReport getSchedulerAppInfo(
      ApplicationAttemptId appAttemptId); @Override ApplicationResourceUsageReport getAppResourceUsageReport(
      ApplicationAttemptId appAttemptId); T getCurrentAttemptForContainer(ContainerId containerId); @Override RMContainer getRMContainer(ContainerId containerId); @Override SchedulerNodeReport getNodeReport(NodeId nodeId); @Override String moveApplication(ApplicationId appId, String newQueue); @Override void preValidateMoveApplication(ApplicationId appId,
      String newQueue); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); void recoverContainersOnNode(List<NMContainerStatus> containerReports,
      RMNode nm); @VisibleForTesting void clearPendingContainerCache(); @VisibleForTesting @Private // clean up a completed container void completedContainer(RMContainer rmContainer,
      ContainerStatus containerStatus, RMContainerEventType event); @Override N getSchedulerNode(NodeId nodeId); @Override void moveAllApps(String sourceQueue, String destQueue); @Override void killAllAppsInQueue(String queueName); void updateNodeResource(RMNode nm,
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); boolean placementConstraintEnabled(); @Override List<ResourceRequest> getPendingResourceRequestsForAttempt(
      ApplicationAttemptId attemptId); @Override List<SchedulingRequest> getPendingSchedulingRequestsForAttempt(
      ApplicationAttemptId attemptId); @Override Priority checkAndGetApplicationPriority(
          Priority priorityRequestedByApp, UserGroupInformation user,
          String queuePath, ApplicationId applicationId); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); @Override Priority getMaxClusterLevelAppPriority(); @Override void setClusterMaxPriority(Configuration conf); ActivitiesManager getActivitiesManager(); Clock getClock(); @VisibleForTesting void setClock(Clock clock); @Lock(Lock.NoLock.class) SchedulerNode getNode(NodeId nodeId); @Override Resource getNormalizedResource(Resource requestedResource,
                                        Resource maxResourceCapability); @Override List<NodeId> getNodeIds(String resourceName); void asyncContainerRelease(RMContainer container); Resource getMinimumAllocation(); Resource getMaximumAllocation(); @Override long checkAndGetApplicationLifetime(String queueName, long lifetime); @Override long getMaximumApplicationLifetime(String queueName); @VisibleForTesting abstract void killContainer(RMContainer container); @VisibleForTesting void update(); @Override void reinitialize(Configuration conf, RMContext rmContext); @Override boolean attemptAllocationOnNode(SchedulerApplicationAttempt appAttempt,
      SchedulingRequest schedulingRequest, SchedulerNode schedulerNode); @Override void resetSchedulerMetrics();  }
@Test(timeout = 60000) public void testNodemanagerReconnect() throws Exception { Configuration conf = getConf(); MockRM rm = new MockRM(conf); try { rm.start(); DrainDispatcher privateDispatcher = new DrainDispatcher(); privateDispatcher.disableExitOnDispatchException(); SleepHandler sleepHandler = new SleepHandler(); ResourceTrackerService privateResourceTrackerService = getPrivateResourceTrackerService(privateDispatcher, rm, sleepHandler); String hostname1 = "localhost1"; Resource capability = BuilderUtils.newResource(4096, 4); RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null); RegisterNodeManagerRequest request1 = recordFactory.newRecordInstance(RegisterNodeManagerRequest.class); NodeId nodeId1 = NodeId.newInstance(hostname1, 0); NodeStatus mockNodeStatus = createMockNodeStatus(); request1.setNodeId(nodeId1); request1.setHttpPort(0); request1.setResource(capability); request1.setNodeStatus(mockNodeStatus); privateResourceTrackerService.registerNodeManager(request1); privateDispatcher.await(); Resource clusterResource = rm.getResourceScheduler().getClusterResource(); Assert.assertEquals("Initial cluster resources don't match", capability, clusterResource); Resource newCapability = BuilderUtils.newResource(1024, 1); RegisterNodeManagerRequest request2 = recordFactory.newRecordInstance(RegisterNodeManagerRequest.class); request2.setNodeId(nodeId1); request2.setHttpPort(0); request2.setResource(newCapability); sleepHandler.sleepFlag = true; privateResourceTrackerService.registerNodeManager(request2); privateDispatcher.await(); Assert.assertEquals("Cluster resources don't match", newCapability, rm.getResourceScheduler().getClusterResource()); privateResourceTrackerService.stop(); } finally { rm.stop(); } }