@Test
public void testGracefulDecommissionWithApp() throws Exception {

  Configuration conf = new Configuration();
  conf.set(YarnConfiguration.RM_NODES_EXCLUDE_FILE_PATH, hostFile
      .getAbsolutePath());
  writeToHostsFile("");
  rm = new MockRM(conf);
  rm.start();
  MockNM nm1 = rm.registerNode("host1:1234", 10240);
  MockNM nm2 = rm.registerNode("host2:5678", 20480);
  MockNM nm3 = rm.registerNode("host3:4433", 10240);
  NodeId id1 = nm1.getNodeId();
  NodeId id3 = nm3.getNodeId();
  rm.waitForState(id1, NodeState.RUNNING);
  rm.waitForState(id3, NodeState.RUNNING);
  // Create an app and launch two containers on host1.
  RMApp app = MockRMAppSubmitter.submitWithMemory(2000, rm);
  MockAM am = MockRM.launchAndRegisterAM(app, rm, nm1);
  ApplicationAttemptId aaid = app.getCurrentAppAttempt().getAppAttemptId();
  nm1.nodeHeartbeat(aaid, 2, ContainerState.RUNNING);
  nm3.nodeHeartbeat(true);
  // Graceful decommission host1 and host3
  writeToHostsFile("host1", "host3");
  rm.getNodesListManager().refreshNodes(conf, true);
  rm.waitForState(id1, NodeState.DECOMMISSIONING);
  rm.waitForState(id3, NodeState.DECOMMISSIONING);
  // host1 should be DECOMMISSIONING due to running containers.
  // host3 should become DECOMMISSIONED.
  nm1.nodeHeartbeat(true);
  nm3.nodeHeartbeat(true);
  rm.waitForState(id1, NodeState.DECOMMISSIONING);
  rm.waitForState(id3, NodeState.DECOMMISSIONED);
  nm1.nodeHeartbeat(aaid, 2, ContainerState.RUNNING);
  // Complete containers on host1.
  // Since the app is still RUNNING, expect NodeAction.NORMAL.
  NodeHeartbeatResponse nodeHeartbeat1 =
      nm1.nodeHeartbeat(aaid, 2, ContainerState.COMPLETE);
  Assert.assertEquals(NodeAction.NORMAL, nodeHeartbeat1.getNodeAction());
  // Finish the app and verified DECOMMISSIONED.
  MockRM.finishAMAndVerifyAppState(app, rm, nm1, am);
  rm.waitForState(app.getApplicationId(), RMAppState.FINISHED);
  nodeHeartbeat1 = nm1.nodeHeartbeat(aaid, 2, ContainerState.COMPLETE);
  Assert.assertEquals(NodeAction.SHUTDOWN, nodeHeartbeat1.getNodeAction());
  rm.waitForState(id1, NodeState.DECOMMISSIONED);
}