@InterfaceAudience.Private
  @VisibleForTesting
  void cleanLogs(Path dirpath, long retainMillis)
      throws IOException {
    long now = Time.now();
    RemoteIterator<FileStatus> iter = list(dirpath);
    while (iter.hasNext()) {
      FileStatus stat = iter.next();
      if (isValidClusterTimeStampDir(stat)) {
        Path clusterTimeStampPath = stat.getPath();
        MutableBoolean appLogDirPresent = new MutableBoolean(false);
        cleanAppLogDir(clusterTimeStampPath, retainMillis, appLogDirPresent);
        if (appLogDirPresent.isFalse() &&
            (now - stat.getModificationTime() > retainMillis)) {
          deleteDir(clusterTimeStampPath);
        }
      }
    }
  }
@Test
  void testCleanBuckets() throws Exception {
    // ClusterTimeStampDir with App Log Dirs
    Path clusterTimeStampDir1 = new Path(testDoneDirPath,
        Long.toString(sampleAppIds.get(0).getClusterTimestamp()));
    Path appDir1 = new Path(new Path(new Path(
        clusterTimeStampDir1, "0000"), "000"), sampleAppIds.get(0).toString());
    Path appDir2 = new Path(new Path(new Path(
        clusterTimeStampDir1, "0000"), "001"), sampleAppIds.get(1).toString());
    Path appDir3 = new Path(new Path(new Path(
        clusterTimeStampDir1, "0000"), "002"), sampleAppIds.get(2).toString());
    Path appDir4 = new Path(new Path(new Path(
        clusterTimeStampDir1, "0001"), "000"), sampleAppIds.get(3).toString());

    // ClusterTimeStampDir with no App Log Dirs
    Path clusterTimeStampDir2 = new Path(testDoneDirPath, "1235");

    // Irrevelant ClusterTimeStampDir
    Path clusterTimeStampDir3 = new Path(testDoneDirPath, "irrevelant");
    Path appDir5 = new Path(new Path(new Path(
        clusterTimeStampDir3, "0000"), "000"), sampleAppIds.get(4).toString());

    fs.mkdirs(appDir1);
    fs.mkdirs(appDir2);
    fs.mkdirs(appDir3);
    fs.mkdirs(appDir4);
    fs.mkdirs(clusterTimeStampDir2);
    fs.mkdirs(appDir5);

    Thread.sleep(2000);

    store.cleanLogs(testDoneDirPath, 1000);

    // ClusterTimeStampDir will be removed only if no App Log Dir Present
    assertTrue(fs.exists(clusterTimeStampDir1));
    assertFalse(fs.exists(appDir1));
    assertFalse(fs.exists(appDir2));
    assertFalse(fs.exists(appDir3));
    assertFalse(fs.exists(appDir4));
    assertFalse(fs.exists(clusterTimeStampDir2));
    assertTrue(fs.exists(appDir5));

    store.cleanLogs(testDoneDirPath, 1000);
    assertFalse(fs.exists(clusterTimeStampDir1));
  }