@Test
public void testReconnectNode() throws Exception {

  rm = new MockRM() {
    @Override
    protected EventHandler<SchedulerEvent> createSchedulerEventDispatcher() {
      return new EventDispatcher<SchedulerEvent>(this.scheduler,
          this.scheduler.getClass().getName()) {
        @Override
        public void handle(SchedulerEvent event) {
          scheduler.handle(event);
        }
      };
    }
  };
  rm.start();
  MockNM nm1 = rm.registerNode("host1:1234", 5120);
  MockNM nm2 = rm.registerNode("host2:5678", 5120);
  nm1.nodeHeartbeat(true);
  nm2.nodeHeartbeat(false);
  rm.drainEvents();
  checkUnhealthyNMCount(rm, nm2, true, 1);
  final int expectedNMs = ClusterMetrics.getMetrics().getNumActiveNMs();
  QueueMetrics metrics = rm.getResourceScheduler().getRootQueueMetrics();
  // TODO Metrics incorrect in case of the FifoScheduler
  Assert.assertEquals(5120, metrics.getAvailableMB());
  // reconnect of healthy node
  nm1 = rm.registerNode("host1:1234", 5120);
  NodeHeartbeatResponse response = nm1.nodeHeartbeat(true);
  Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction()));
  rm.drainEvents();
  Assert.assertEquals(expectedNMs, ClusterMetrics.getMetrics().getNumActiveNMs());
  checkUnhealthyNMCount(rm, nm2, true, 1);
  // reconnect of unhealthy node
  nm2 = rm.registerNode("host2:5678", 5120);
  response = nm2.nodeHeartbeat(false);
  Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction()));
  rm.drainEvents();
  Assert.assertEquals(expectedNMs, ClusterMetrics.getMetrics().getNumActiveNMs());
  checkUnhealthyNMCount(rm, nm2, true, 1);
  
  // unhealthy node changed back to healthy
  nm2 = rm.registerNode("host2:5678", 5120);
  response = nm2.nodeHeartbeat(true);
  response = nm2.nodeHeartbeat(true);
  rm.drainEvents();
  Assert.assertEquals(5120 + 5120, metrics.getAvailableMB());
  // reconnect of node with changed capability
  nm1 = rm.registerNode("host2:5678", 10240);
  response = nm1.nodeHeartbeat(true);
  rm.drainEvents();
  Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction()));
  Assert.assertEquals(5120 + 10240, metrics.getAvailableMB());
  // reconnect of node with changed capability and running applications
  List<ApplicationId> runningApps = new ArrayList<ApplicationId>();
  runningApps.add(ApplicationId.newInstance(1, 0));
  nm1 = rm.registerNode("host2:5678", 15360, 2, runningApps);
  response = nm1.nodeHeartbeat(true);
  rm.drainEvents();
  Assert.assertTrue(NodeAction.NORMAL.equals(response.getNodeAction()));
  Assert.assertEquals(5120 + 15360, metrics.getAvailableMB());
  
  // reconnect healthy node changing http port
  nm1 = new MockNM("host1:1234", 5120, rm.getResourceTrackerService());
  nm1.setHttpPort(3);
  nm1.registerNode();
  response = nm1.nodeHeartbeat(true);
  response = nm1.nodeHeartbeat(true);
  rm.drainEvents();
  RMNode rmNode = rm.getRMContext().getRMNodes().get(nm1.getNodeId());
  Assert.assertEquals(3, rmNode.getHttpPort());
  Assert.assertEquals(5120, rmNode.getTotalCapability().getMemorySize());
  Assert.assertEquals(5120 + 15360, metrics.getAvailableMB());
}