@Override
  public Resource getClusterResource() {
    return nodeTracker.getClusterCapacity();
  }
@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);

      // Register node1
      String hostname1 = "localhost1";
      Resource capability = Resources.createResource(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 = Resources.createResource(1024);
      RegisterNodeManagerRequest request2 =
          recordFactory.newRecordInstance(RegisterNodeManagerRequest.class);
      request2.setNodeId(nodeId1);
      request2.setHttpPort(0);
      request2.setResource(newCapability);
      // hold up the disaptcher and register the same node with lower capability
      sleepHandler.sleepFlag = true;
      privateResourceTrackerService.registerNodeManager(request2);
      privateDispatcher.await();
      Assert.assertEquals("Cluster resources don't match", newCapability,
          rm.getResourceScheduler().getClusterResource());
      privateResourceTrackerService.stop();
    } finally {
      rm.stop();
    }
  }