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

  String remoteLogRootDir = "target/logs/";
  String rootLogDir = "target/LocalLogs";
  String testUser = "test";
  UserGroupInformation testUgi = UserGroupInformation
      .createRemoteUser(testUser);
  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);
  ContainerId containerId = ContainerId
      .newContainerId(appAttemptId, 1);
  NodeId nodeId = NodeId.newInstance("localhost", 1234);
  try {
    Path rootLogDirPath = new Path(rootLogDir);
    if (fs.exists(rootLogDirPath)) {
      fs.delete(rootLogDirPath, true);
    }
    assertTrue(fs.mkdirs(rootLogDirPath));
    // create local app dir for app
    final 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 for app
    createContainerLogInLocalDir(appLogsDir, containerId, fs, logTypes,
        Collections.emptyList());
    // create the remote app dir for app but for a different user testUser
    Path path = new Path(remoteLogRootDir + testUser +
        "/bucket-logs-tfile/0001/" + appId);
    if (fs.exists(path)) {
      fs.delete(path, true);
    }
    assertTrue(fs.mkdirs(path));
    // upload container logs for app into remote dir
    uploadContainerLogIntoRemoteDir(testUgi, conf, rootLogDirs,
        nodeId, containerId, path, fs);
    YarnClient mockYarnClient = createMockYarnClient(
        YarnApplicationState.FINISHED, testUgi.getShortUserName());
    LogsCLI cli = new LogsCLIForTest(mockYarnClient);
    cli.setConf(conf);
    // Verify that we can get the application logs by specifying
    // a correct appOwner
    int exitCode = cli.run(new String[] {
        "-applicationId", appId.toString(),
        "-appOwner", testUser});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId, "syslog")));
    sysOutStream.reset();
    // Verify that we can not get the application logs
    // if an invalid user is specified
    exitCode = cli.run(new String[] {
        "-applicationId", appId.toString(),
        "-appOwner", "invalid"});
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains("Can not find the logs "
        + "for the application: " + appId.toString()));
    sysErrStream.reset();
    // Verify that we do not specify appOwner, and can not
    // get appReport from RM, we still can figure out the appOwner
    // and can get app logs successfully.
    YarnClient mockYarnClient2 = createMockYarnClientUnknownApp();
    cli = new LogsCLIForTest(mockYarnClient2);
    cli.setConf(conf);
    exitCode = cli.run(new String[] {
        "-applicationId", appId.toString()});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId, "syslog")));
    sysOutStream.reset();
    // Verify appOwner guessed correctly with older log dir dtructure
    path = new Path(remoteLogRootDir + testUser + "/logs/" + appId);
    if (fs.exists(path)) {
      fs.delete(path, true);
    }
    assertTrue(fs.mkdirs(path));
    uploadContainerLogIntoRemoteDir(testUgi, conf, rootLogDirs,
        nodeId, containerId, path, fs);
    exitCode = cli.run(new String[] {
        "-applicationId", appId.toString()});
    assertTrue(exitCode == 0);
    assertTrue(sysOutStream.toString().contains(
        logMessage(containerId, "syslog")));
    sysOutStream.reset();
    // Verify that we could get the err message "Can not find the appOwner"
    // if we do not specify the appOwner, can not get appReport, and
    // the app does not exist in remote dir.
    ApplicationId appId2 = ApplicationId.newInstance(
        System.currentTimeMillis(), 2);
    exitCode = cli.run(new String[] {
        "-applicationId", appId2.toString()});
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Can not find the appOwner"));
    sysErrStream.reset();
    // Verify that we could not get appOwner
    // because we don't have file-system permissions
    ApplicationId appTest = ApplicationId.newInstance(
        System.currentTimeMillis(), 1000);
    String priorityUser = "priority";
    Path pathWithoutPerm = new Path(remoteLogRootDir + priorityUser
        + "/bucket-logs-tfile/1000/" + appTest);
    if (fs.exists(pathWithoutPerm)) {
      fs.delete(pathWithoutPerm, true);
    }
    // The user will not have read permission for this directory.
    // To mimic the scenario that the user can not get file status
    FsPermission permission = FsPermission
        .createImmutable((short) 01300);
    assertTrue(fs.mkdirs(pathWithoutPerm, permission));
    exitCode = cli.run(new String[] {
        "-applicationId", appTest.toString()});
    assertTrue(exitCode == -1);
    assertTrue(sysErrStream.toString().contains(
        "Can not find the logs for the application: "
        + appTest.toString()));
    sysErrStream.reset();
  } finally {
    fs.delete(new Path(remoteLogRootDir), true);
    fs.delete(new Path(rootLogDir), true);
  }
}