@Test(timeout = 60000) public void testNodemanagerReconnect() throws Exception { configureScheduler(); Configuration conf = getConf(); MockRM rm = new MockRM(conf); try { rm.start(); conf.setBoolean(Dispatcher.DISPATCHER_EXIT_ON_ERROR_KEY, false); DrainDispatcher privateDispatcher = new DrainDispatcher(); 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); request1.setNodeId(nodeId1); request1.setHttpPort(0); request1.setResource(capability); 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(); } }
@Override public Resource getClusterResource() { return clusterResource; }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { @Override public Resource getClusterResource() { return clusterResource; } }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { @Override public Resource getClusterResource() { return clusterResource; } AbstractYarnScheduler(String name); }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { @Override public Resource getClusterResource() { return clusterResource; } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); synchronized List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); 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); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm); SchedulerNode getSchedulerNode(NodeId nodeId); @Override synchronized void moveAllApps(String sourceQueue, String destQueue); @Override synchronized void killAllAppsInQueue(String queueName); synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); }
AbstractYarnScheduler extends AbstractService implements ResourceScheduler { @Override public Resource getClusterResource() { return clusterResource; } AbstractYarnScheduler(String name); @Override void serviceInit(Configuration conf); synchronized List<Container> getTransferredContainers(
      ApplicationAttemptId currentAttempt); Map<ApplicationId, SchedulerApplication<T>> getSchedulerApplications(); @Override Resource getClusterResource(); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); 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); void removeQueue(String queueName); @Override void addQueue(Queue newQueue); @Override void setEntitlement(String queue, QueueEntitlement entitlement); synchronized void recoverContainersOnNode(
      List<NMContainerStatus> containerReports, RMNode nm); SchedulerNode getSchedulerNode(NodeId nodeId); @Override synchronized void moveAllApps(String sourceQueue, String destQueue); @Override synchronized void killAllAppsInQueue(String queueName); synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues();  }