@Test (timeout = 15000)
public void testSaveContainerLogsLocally() throws Exception {

  String remoteLogRootDir = "target/logs/";
  String rootLogDir = "target/LocalLogs";
  String localDir = "target/SaveLogs";
  Path localPath = new Path(localDir);
  conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
  conf
      .set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir);
  conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
  conf.set(YarnConfiguration.YARN_ADMIN_ACL, "admin");
  FileSystem fs = FileSystem.get(conf);
  ApplicationId appId = ApplicationId.newInstance(0, 1);
  ApplicationAttemptId appAttemptId =
      ApplicationAttemptId.newInstance(appId, 1);
  List<ContainerId> containerIds = new ArrayList<ContainerId>();
  ContainerId containerId1 = ContainerId.newContainerId(
      appAttemptId, 1);
  ContainerId containerId2 = ContainerId.newContainerId(
      appAttemptId, 2);
  containerIds.add(containerId1);
  containerIds.add(containerId2);
  List<NodeId> nodeIds = new ArrayList<NodeId>();
  NodeId nodeId = NodeId.newInstance("localhost", 1234);
  NodeId nodeId2 = NodeId.newInstance("test", 4567);
  nodeIds.add(nodeId);
  nodeIds.add(nodeId2);
  try {
    createContainerLogs(conf, remoteLogRootDir, rootLogDir, fs,
        appId, containerIds, nodeIds);
    YarnClient mockYarnClient =
        createMockYarnClient(YarnApplicationState.FINISHED,
        UserGroupInformation.getCurrentUser().getShortUserName());
    LogsCLI cli = new LogsCLIForTest(mockYarnClient);
    cli.setConf(conf);
    int exitCode = cli.run(new String[] {"-applicationId",
        appId.toString(),
        "-out" , localPath.toString()});
    assertTrue(exitCode == 0);
    // make sure we created a dir named as node id
    FileStatus[] nodeDir = fs.listStatus(localPath);
    Arrays.sort(nodeDir);
    assertTrue(nodeDir.length == 2);
    assertTrue(nodeDir[0].getPath().getName().contains(
        LogAggregationUtils.getNodeString(nodeId)));
    assertTrue(nodeDir[1].getPath().getName().contains(
        LogAggregationUtils.getNodeString(nodeId2)));
    FileStatus[] container1Dir = fs.listStatus(nodeDir[0].getPath());
    assertTrue(container1Dir.length == 1);
    assertTrue(container1Dir[0].getPath().getName().equals(
        containerId1.toString()));
    String container1= readContainerContent(container1Dir[0].getPath(), fs);
    assertTrue(container1.contains(logMessage(containerId1, "syslog")));
    FileStatus[] container2Dir = fs.listStatus(nodeDir[1].getPath());
    assertTrue(container2Dir.length == 1);
    assertTrue(container2Dir[0].getPath().getName().equals(
        containerId2.toString()));
    String container2= readContainerContent(container2Dir[0].getPath(), fs);
    assertTrue(container2.contains(logMessage(containerId2, "syslog")));
  } finally {
    fs.delete(new Path(remoteLogRootDir), true);
    fs.delete(new Path(rootLogDir), true);
    fs.delete(localPath, true);
  }
}