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

  String remoteLogRootDir = "target/logs/";
  Configuration configuration = new Configuration();
  configuration.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
  configuration
    .set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteLogRootDir);
  configuration.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
  configuration.set(YarnConfiguration.YARN_ADMIN_ACL, "admin");
  FileSystem fs = FileSystem.get(configuration);
  UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
  ApplicationId appId = ApplicationIdPBImpl.newInstance(0, 1);
  ApplicationAttemptId appAttemptId =
      ApplicationAttemptIdPBImpl.newInstance(appId, 1);
  ContainerId containerId0 = ContainerIdPBImpl.newContainerId(appAttemptId, 0);
  ContainerId containerId1 = ContainerIdPBImpl.newContainerId(appAttemptId, 1);
  ContainerId containerId2 = ContainerIdPBImpl.newContainerId(appAttemptId, 2);
  NodeId nodeId = NodeId.newInstance("localhost", 1234);
  // create local logs
  String rootLogDir = "target/LocalLogs";
  Path rootLogDirPath = new Path(rootLogDir);
  if (fs.exists(rootLogDirPath)) {
    fs.delete(rootLogDirPath, true);
  }
  assertTrue(fs.mkdirs(rootLogDirPath));
  Path appLogsDir = new Path(rootLogDirPath, appId.toString());
  if (fs.exists(appLogsDir)) {
    fs.delete(appLogsDir, true);
  }
  assertTrue(fs.mkdirs(appLogsDir));
  List<String> rootLogDirs = Arrays.asList(rootLogDir);
  // create container logs in localLogDir
  createContainerLogInLocalDir(appLogsDir, containerId1, fs);
  createContainerLogInLocalDir(appLogsDir, containerId2, fs);
  Path path =
      new Path(remoteLogRootDir + ugi.getShortUserName()
          + "/logs/application_0_0001");
  if (fs.exists(path)) {
    fs.delete(path, true);
  }
  assertTrue(fs.mkdirs(path));
  // upload container logs into remote directory
  // the first two logs is empty. When we try to read first two logs,
  // we will meet EOF exception, but it will not impact other logs.
  // Other logs should be read successfully.
  uploadEmptyContainerLogIntoRemoteDir(ugi, configuration, rootLogDirs, nodeId,
    containerId0, path, fs);
  uploadEmptyContainerLogIntoRemoteDir(ugi, configuration, rootLogDirs, nodeId,
    containerId1, path, fs);
  uploadContainerLogIntoRemoteDir(ugi, configuration, rootLogDirs, nodeId,
    containerId1, path, fs);
  uploadContainerLogIntoRemoteDir(ugi, configuration, rootLogDirs, nodeId,
    containerId2, path, fs);
  YarnClient mockYarnClient =
      createMockYarnClient(YarnApplicationState.FINISHED);
  LogsCLI cli = new LogsCLIForTest(mockYarnClient);
  cli.setConf(configuration);
  int exitCode = cli.run(new String[] { "-applicationId", appId.toString() });
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
    "Hello container_0_0001_01_000001!"));
  assertTrue(sysOutStream.toString().contains(
    "Hello container_0_0001_01_000002!"));
  sysOutStream.reset();
  // uploaded two logs for container1. The first log is empty.
  // The second one is not empty.
  // We can still successfully read logs for container1.
  exitCode =
      cli.run(new String[] { "-applicationId", appId.toString(),
          "-nodeAddress", nodeId.toString(), "-containerId",
          containerId1.toString() });
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      "Hello container_0_0001_01_000001!"));
  assertTrue(sysOutStream.toString().contains("Log Upload Time"));
  assertTrue(!sysOutStream.toString().contains(
    "Logs for container " + containerId1.toString()
        + " are not present in this log-file."));
  sysOutStream.reset();
  // Uploaded the empty log for container0.
  // We should see the message showing the log for container0
  // are not present.
  exitCode =
      cli.run(new String[] { "-applicationId", appId.toString(),
          "-nodeAddress", nodeId.toString(), "-containerId",
          containerId0.toString() });
  assertTrue(exitCode == -1);
  assertTrue(sysOutStream.toString().contains(
    "Logs for container " + containerId0.toString()
        + " are not present in this log-file."));
  fs.delete(new Path(remoteLogRootDir), true);
  fs.delete(new Path(rootLogDir), true);
}