@VisibleForTesting
  void checkFilesAndSeedApps(FileSystem fs, Path remoteRootLogDir,
       String suffix, Path workingDir) throws IOException {
    for (RemoteIterator<FileStatus> userIt =
         fs.listStatusIterator(remoteRootLogDir); userIt.hasNext();) {
      Path userLogPath = userIt.next().getPath();
      try {
        for (RemoteIterator<FileStatus> appIt =
             fs.listStatusIterator(new Path(userLogPath, suffix));
             appIt.hasNext();) {
          Path appLogPath = appIt.next().getPath();
          try {
            FileStatus[] files = fs.listStatus(appLogPath);
            if (files.length >= minNumLogFiles) {
              boolean eligible = true;
              long totalFileSize = 0L;
              for (FileStatus file : files) {
                if (file.getPath().getName().equals(appLogPath.getName()
                    + ".har")) {
                  eligible = false;
                  if (verbose) {
                    LOG.info("Skipping " + appLogPath.getName() +
                        " due to existing .har file");
                  }
                  break;
                }
                totalFileSize += file.getLen();
                if (totalFileSize > maxTotalLogsSize) {
                  eligible = false;
                  if (verbose) {
                    LOG.info("Skipping " + appLogPath.getName() + " due to " +
                        "total file size being too large (" + totalFileSize +
                        " > " + maxTotalLogsSize + ")");
                  }
                  break;
                }
              }
              if (eligible) {
                if (verbose) {
                  LOG.info("Adding " + appLogPath.getName() + " for user " +
                      userLogPath.getName());
                }
                AppInfo context = new AppInfo();
                context.setAppId(appLogPath.getName());
                context.setUser(userLogPath.getName());
                context.setSuffix(suffix);
                context.setRemoteRootLogDir(remoteRootLogDir);
                context.setWorkingDir(workingDir);
                eligibleApplications.add(context);
              }
            } else {
              if (verbose) {
                LOG.info("Skipping " + appLogPath.getName() + " due to not " +
                    "having enough log files (" + files.length + " < " +
                    minNumLogFiles + ")");
              }
            }
          } catch (IOException ioe) {
            // Ignore any apps we can't read
            if (verbose) {
              LOG.info("Skipping logs under " + appLogPath + " due to " +
                  ioe.getMessage());
            }
          }
        }
      } catch (IOException ioe) {
        // Ignore any apps we can't read
        if (verbose) {
          LOG.info("Skipping all logs under " + userLogPath + " due to " +
              ioe.getMessage());
        }
      }
    }
  }
@Test(timeout = 10000)
  public void testCheckFilesAndSeedApps() throws Exception {
    Configuration conf = new Configuration();
    HadoopArchiveLogs hal = new HadoopArchiveLogs(conf);
    FileSystem fs = FileSystem.getLocal(conf);
    Path rootLogDir = new Path("target", "logs");
    String suffix = "logs";
    Path logDir = new Path(rootLogDir, new Path(USER, suffix));
    fs.delete(logDir, true);
    Assert.assertFalse(fs.exists(logDir));
    fs.mkdirs(logDir);

    // no files found
    ApplicationId appId1 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 1);
    Path app1Path = new Path(logDir, appId1.toString());
    fs.mkdirs(app1Path);
    // too few files
    ApplicationId appId2 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 2);
    Path app2Path = new Path(logDir, appId2.toString());
    fs.mkdirs(app2Path);
    createFile(fs, new Path(app2Path, "file1"), 1);
    hal.minNumLogFiles = 2;
    // too large
    ApplicationId appId3 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 3);
    Path app3Path = new Path(logDir, appId3.toString());
    fs.mkdirs(app3Path);
    createFile(fs, new Path(app3Path, "file1"), 2);
    createFile(fs, new Path(app3Path, "file2"), 5);
    hal.maxTotalLogsSize = FILE_SIZE_INCREMENT * 6;
    // has har already
    ApplicationId appId4 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 4);
    Path app4Path = new Path(logDir, appId4.toString());
    fs.mkdirs(app4Path);
    createFile(fs, new Path(app4Path, appId4 + ".har"), 1);
    // just right
    ApplicationId appId5 = ApplicationId.newInstance(CLUSTER_TIMESTAMP, 5);
    Path app5Path = new Path(logDir, appId5.toString());
    fs.mkdirs(app5Path);
    createFile(fs, new Path(app5Path, "file1"), 2);
    createFile(fs, new Path(app5Path, "file2"), 3);

    Assert.assertEquals(0, hal.eligibleApplications.size());
    hal.checkFilesAndSeedApps(fs, rootLogDir, suffix, new Path(rootLogDir,
        "archive-logs-work"));
    Assert.assertEquals(1, hal.eligibleApplications.size());
    Assert.assertEquals(appId5.toString(),
        hal.eligibleApplications.iterator().next().getAppId());
  }