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

  String remoteLogRootDir = "target/logs/";
  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);
  UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
  ApplicationId appId = ApplicationId.newInstance(0, 1);
  ApplicationAttemptId appAttemptId1 =
      ApplicationAttemptId.newInstance(appId, 1);
  ApplicationAttemptId appAttemptId2 =
          ApplicationAttemptId.newInstance(appId, 2);
  ContainerId containerId0 = ContainerId.newContainerId(appAttemptId1, 0);
  ContainerId containerId1 = ContainerId.newContainerId(appAttemptId1, 1);
  ContainerId containerId2 = ContainerId.newContainerId(appAttemptId1, 2);
  ContainerId containerId3 = ContainerId.newContainerId(appAttemptId2, 3);
  final 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);
  List<String> logTypes = new ArrayList<String>();
  logTypes.add("syslog");
  // create container logs in localLogDir
  createContainerLogInLocalDir(appLogsDir, containerId1, fs, logTypes,
      ImmutableList.of("empty"));
  createContainerLogInLocalDir(appLogsDir, containerId2, fs, logTypes,
      Collections.emptyList());
  // create two logs for container3 in localLogDir
  logTypes.add("stdout");
  logTypes.add("stdout1234");
  createContainerLogInLocalDir(appLogsDir, containerId3, fs, logTypes,
      Collections.emptyList());
  Path path =
      new Path(remoteLogRootDir + ugi.getShortUserName()
      + "/bucket-logs-tfile/0001/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, conf, rootLogDirs, nodeId,
    containerId0, path, fs);
  uploadEmptyContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
    containerId1, path, fs);
  uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
    containerId1, path, fs);
  uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
    containerId2, path, fs);
  uploadContainerLogIntoRemoteDir(ugi, conf, rootLogDirs, nodeId,
    containerId3, path, fs);
  YarnClient mockYarnClient =
      createMockYarnClient(
          YarnApplicationState.FINISHED, ugi.getShortUserName());
  LogsCLI cli = new LogsCLIForTest(mockYarnClient) {
    @Override
    public ContainerReport getContainerReport(String containerIdStr)
        throws YarnException, IOException {
      ContainerReport mockReport = mock(ContainerReport.class);
      doReturn(nodeId).when(mockReport).getAssignedNode();
      doReturn("http://localhost:2345").when(mockReport).getNodeHttpAddress();
      return mockReport;
    }
  };
  cli.setConf(conf);
  int exitCode = cli.run(new String[] { "-applicationId", appId.toString() });
  LOG.info(sysOutStream.toString());
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertTrue(sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  // Check fetching data for application attempt with applicationId defined
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-applicationAttemptId", appAttemptId1.toString()});
  LOG.info(sysOutStream.toString());
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
          logMessage(containerId1, "syslog")));
  assertTrue(sysOutStream.toString().contains(
          logMessage(containerId2, "syslog")));
  assertTrue(sysOutStream.toString().contains(
          logMessage(containerId3, "syslog")));
  assertFalse(sysOutStream.toString().contains(
          logMessage(containerId3, "stdout")));
  assertFalse(sysOutStream.toString().contains(
          logMessage(containerId3, "stdout1234")));
  assertTrue(sysOutStream.toString().contains(
          createEmptyLog("empty")));
  sysOutStream.reset();
  // Check fetching data for application attempt without application defined
  exitCode = cli.run(new String[] {
      "-applicationAttemptId", appAttemptId1.toString()});
  LOG.info(sysOutStream.toString());
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
          logMessage(containerId1, "syslog")));
  assertTrue(sysOutStream.toString().contains(
          logMessage(containerId2, "syslog")));
  assertTrue(sysOutStream.toString().contains(
          logMessage(containerId3, "syslog")));
  assertFalse(sysOutStream.toString().contains(
          logMessage(containerId3, "stdout")));
  assertFalse(sysOutStream.toString().contains(
          logMessage(containerId3, "stdout1234")));
  assertTrue(sysOutStream.toString().contains(
          createEmptyLog("empty")));
  sysOutStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files_pattern", ".*"});
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertTrue(sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files", "*"});
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertTrue(sysOutStream.toString().contains(
      createEmptyLog("empty")));
  int fullSize = sysOutStream.toByteArray().length;
  sysOutStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files", "stdout"});
  assertTrue(exitCode == 0);
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertFalse(sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  // Check backward compatibility for -logFiles
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-logFiles", "stdout"});
  assertTrue("Failed with -logFiles", exitCode == 0);
  assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue("Failed with -logFiles", sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertFalse("Failed with -logFiles", sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  // Check -log_files supercedes -logFiles
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files", "stdout", "-logFiles", "syslog"});
  assertTrue("Failed with -logFiles and -log_files", exitCode == 0);
  assertFalse("Failed with -logFiles and -log_files",
      sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertFalse("Failed with -logFiles and -log_files",
      sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertFalse("Failed with -logFiles and -log_files",
      sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue("Failed with -logFiles and -log_files",
      sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertFalse("Failed with -logFiles and -log_files",
      sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertFalse("Failed with -logFiles and -log_files",
      sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files_pattern", "std*"});
  assertTrue(exitCode == 0);
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId1, "syslog")));
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId2, "syslog")));
  assertFalse(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout1234")));
  assertFalse(sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files", "123"});
  assertTrue(exitCode == -1);
  assertTrue(sysErrStream.toString().contains(
      "Can not find any log file matching the pattern: [123] "
      + "for the application: " + appId.toString()));
  sysErrStream.reset();
  // specify the bytes which is larger than the actual file size,
  // we would get the full logs
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-log_files", "*", "-size", "10000" });
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toByteArray().length == fullSize);
  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(
      logMessage(containerId1, "syslog")));
  assertTrue(sysOutStream.toString().contains("LogLastModifiedTime"));
  assertTrue(!sysOutStream.toString().contains(
    "Logs for container " + containerId1.toString()
        + " are not present in this log-file."));
  sysOutStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-containerId", containerId3.toString(), "-log_files", "123" });
  assertTrue(exitCode == -1);
  assertTrue(sysErrStream.toString().contains(
      "Can not find any log file matching the pattern: [123] "
      + "for the container: " + containerId3
      + " within the application: " + appId.toString()));
  sysErrStream.reset();
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-containerId", containerId3.toString(), "-log_files", "stdout" });
  assertTrue(exitCode == 0);
  int fullContextSize = sysOutStream.toByteArray().length;
  String fullContext = sysOutStream.toString();
  sysOutStream.reset();
  String logMessage = logMessage(containerId3, "stdout");
  int fileContentSize = logMessage.getBytes().length;
  StringBuilder sb = new StringBuilder();
  String endOfFile = "End of LogType:stdout";
  sb.append("\n" + endOfFile + "\n");
  sb.append(StringUtils.repeat("*", endOfFile.length() + 50)
      + "\n\n");
  int tailContentSize = sb.toString().length();
  // specify how many bytes we should get from logs
  // specify a position number, it would get the first n bytes from
  // container log
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-containerId", containerId3.toString(), "-log_files", "stdout",
      "-size", "5"});
  assertTrue(exitCode == 0);
  Assert.assertEquals(new String(logMessage.getBytes(), 0, 5),
      new String(sysOutStream.toByteArray(),
      (fullContextSize - fileContentSize - tailContentSize), 5));
  sysOutStream.reset();
  // specify how many bytes we should get from an empty log
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-containerId", containerId1.toString(), "-log_files", "empty",
      "-size", "5"});
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      createEmptyLog("empty")));
  sysOutStream.reset();
  // specify a negative number, it would get the last n bytes from
  // container log
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-containerId", containerId3.toString(), "-log_files", "stdout",
      "-size", "-5"});
  assertTrue(exitCode == 0);
  Assert.assertEquals(new String(logMessage.getBytes(),
      logMessage.getBytes().length - 5, 5),
      new String(sysOutStream.toByteArray(),
      (fullContextSize - fileContentSize - tailContentSize), 5));
  sysOutStream.reset();
  long negative = (fullContextSize + 1000) * (-1);
  exitCode = cli.run(new String[] {"-applicationId", appId.toString(),
      "-containerId", containerId3.toString(), "-log_files", "stdout",
      "-size", Long.toString(negative)});
  assertTrue(exitCode == 0);
  Assert.assertEquals(fullContext, sysOutStream.toString());
  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(sysErrStream.toString().contains(
      "Can not find any log file matching the pattern"));
  sysErrStream.reset();
  // uploaded two logs for container3. The first log is named as syslog.
  // The second one is named as stdout.
  exitCode =
      cli.run(new String[] { "-applicationId", appId.toString(),
          "-nodeAddress", nodeId.toString(), "-containerId",
          containerId3.toString() });
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  sysOutStream.reset();
  // set -log_files option as stdout
  // should only print log with the name as stdout
  exitCode =
      cli.run(new String[] { "-applicationId", appId.toString(),
          "-nodeAddress", nodeId.toString(), "-containerId",
          containerId3.toString() , "-log_files", "stdout"});
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(!sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  sysOutStream.reset();
  YarnClient mockYarnClientWithException =
      createMockYarnClientWithException();
  cli = new LogsCLIForTest(mockYarnClientWithException);
  cli.setConf(conf);
  exitCode =
      cli.run(new String[] { "-applicationId", appId.toString(),
          "-containerId", containerId3.toString() });
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(sysOutStream.toString().contains(
      containerId3 + " on " + LogAggregationUtils.getNodeString(nodeId)));
  sysOutStream.reset();
  // The same should also work without the applicationId
  exitCode =
      cli.run(new String[] { "-containerId", containerId3.toString() });
  assertTrue(exitCode == 0);
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "syslog")));
  assertTrue(sysOutStream.toString().contains(
      logMessage(containerId3, "stdout")));
  assertTrue(sysOutStream.toString().contains(
      containerId3 + " on " + LogAggregationUtils.getNodeString(nodeId)));
  sysOutStream.reset();
  exitCode = cli.run(new String[] { "-containerId", "invalid_container" });
  assertTrue(exitCode == -1);
  assertTrue(sysErrStream.toString().contains(
      "Invalid ContainerId specified"));
  sysErrStream.reset();
  fs.delete(new Path(remoteLogRootDir), true);
  fs.delete(new Path(rootLogDir), true);
}