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); void stop(); 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); 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)); rm.sendNodeGracefulDecommission(nm1, YarnConfiguration.DEFAULT_RM_NODE_GRACEFUL_DECOMMISSION_TIMEOUT); rm.waitForState(id1, NodeState.DECOMMISSIONING); 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)); MockRM.finishAMAndVerifyAppState(app, rm, nm1, am); rm.waitForState(app.getApplicationId(), RMAppState.FINISHED); watcher.update(node1, nodeStatus); Assert.assertEquals(DecommissioningNodeStatus.READY, watcher.checkDecommissioningStatus(id1)); }