@Override
  public int run(String[] args) throws Exception {
    try {
      webServiceClient = WebServiceClient.getWebServiceClient().createClient();
      return runCommand(args);
    } finally {
      if (yarnClient != null) {
        yarnClient.close();
      }
      if (webServiceClient != null) {
        webServiceClient.destroy();
      }
    }
  }
@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);
    }
  }