@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);
    // Register node1
    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);
    // 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();
  }
}