public DecommissioningNodesWatcher(RMContext rmContext) {
    this.rmContext = rmContext;
    pollTimer = new Timer(true);
    mclock = new MonotonicClock();
  }
@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));
  }