DecommissioningNodesWatcher { public DecommissioningNodesWatcher(RMContext rmContext) { this.rmContext = rmContext; pollTimer = new Timer(true); mclock = new MonotonicClock(); } DecommissioningNodesWatcher(RMContext rmContext); void init(Configuration conf); synchronized void update(RMNode rmNode, NodeStatus remoteNodeStatus); synchronized void remove(NodeId nodeId); boolean checkReadyToBeDecommissioned(NodeId nodeId); DecommissioningNodeStatus checkDecommissioningStatus(NodeId nodeId);  }
@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); RMNode node1 = rm.getRMContext().getRMNodes().get(nm1.getNodeId()); NodeId id1 = nm1.getNodeId(); rm.waitForState(id1, NodeState.RUNNING); Assert.assertFalse(watcher.checkReadyToBeDecommissioned(id1)); RMApp app = rm.submitApp(2000); MockAM am = MockRM.launchAndRegisterAM(app, rm, nm1); rm.sendNodeGracefulDecommission(nm1, YarnConfiguration.DEFAULT_RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT); rm.waitForState(id1, NodeState.DECOMMISSIONING); watcher.update(node1, createNodeStatus(id1, app, 12)); watcher.update(node1, createNodeStatus(id1, app, 11)); Assert.assertFalse(watcher.checkReadyToBeDecommissioned(id1)); watcher.update(node1, createNodeStatus(id1, app, 1)); Assert.assertEquals(DecommissioningNodeStatus.WAIT_CONTAINER, watcher.checkDecommissioningStatus(id1)); watcher.update(node1, createNodeStatus(id1, app, 0)); Assert.assertEquals(DecommissioningNodeStatus.WAIT_APP, watcher.checkDecommissioningStatus(id1)); MockRM.finishAMAndVerifyAppState(app, rm, nm1, am); rm.waitForState(app.getApplicationId(), RMAppState.FINISHED); Assert.assertEquals(DecommissioningNodeStatus.READY, watcher.checkDecommissioningStatus(id1)); }