@Test
public void testDecommissioningNodesWatcher() throws Exception {

  Configuration conf = new Configuration();
  conf.set(YarnConfiguration.RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT, "40");
  rm = new MockRM(conf);
  rm.start();
  DecommissioningNodesWatcher watcher =
      new DecommissioningNodesWatcher(rm.getRMContext());
  MockNM nm1 = rm.registerNode("host1:1234", 10240);
  RMNodeImpl node1 =
      (RMNodeImpl) rm.getRMContext().getRMNodes().get(nm1.getNodeId());
  NodeId id1 = nm1.getNodeId();
  rm.waitForState(id1, NodeState.RUNNING);
  RMApp app = MockRMAppSubmitter.submitWithMemory(2000, rm);
  MockAM am = MockRM.launchAndRegisterAM(app, rm, nm1);
  NodeStatus nodeStatus = createNodeStatus(id1, app, 3);
  node1.handle(new RMNodeStatusEvent(nm1.getNodeId(), nodeStatus));
  // Setup nm1 as DECOMMISSIONING for DecommissioningNodesWatcher.
  rm.sendNodeGracefulDecommission(nm1,
      YarnConfiguration.DEFAULT_RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT);
  rm.waitForState(id1, NodeState.DECOMMISSIONING);
  // Update status with decreasing number of running containers until 0.
  nodeStatus = createNodeStatus(id1, app, 3);
  node1.handle(new RMNodeStatusEvent(nm1.getNodeId(), nodeStatus));
  watcher.update(node1, nodeStatus);
  nodeStatus = createNodeStatus(id1, app, 2);
  node1.handle(new RMNodeStatusEvent(nm1.getNodeId(), nodeStatus));
  watcher.update(node1, nodeStatus);
  Assert.assertFalse(watcher.checkReadyToBeDecommissioned(id1));
  nodeStatus = createNodeStatus(id1, app, 1);
  node1.handle(new RMNodeStatusEvent(nm1.getNodeId(), nodeStatus));
  watcher.update(node1, nodeStatus);
  Assert.assertEquals(DecommissioningNodeStatus.WAIT_CONTAINER,
      watcher.checkDecommissioningStatus(id1));
  nodeStatus = createNodeStatus(id1, app, 0);
  watcher.update(node1, nodeStatus);
  node1.handle(new RMNodeStatusEvent(nm1.getNodeId(), nodeStatus));
  Assert.assertEquals(DecommissioningNodeStatus.WAIT_APP,
      watcher.checkDecommissioningStatus(id1));
  // Set app to be FINISHED and verified DecommissioningNodeStatus is READY.
  MockRM.finishAMAndVerifyAppState(app, rm, nm1, am);
  rm.waitForState(app.getApplicationId(), RMAppState.FINISHED);
  watcher.update(node1, nodeStatus);
  Assert.assertEquals(DecommissioningNodeStatus.READY,
      watcher.checkDecommissioningStatus(id1));
}