@Test
public void testCleanLogs() throws Exception {

  // Create test dirs and files
  // Irrelevant file, should not be reclaimed
  String appDirName = mainTestAppId.toString();
  String attemptDirName = ApplicationAttemptId.appAttemptIdStrPrefix
      + appDirName + "_1";
  Path irrelevantFilePath = new Path(
          testDoneDirPath, "irrelevant.log");
  FSDataOutputStream stream = fs.create(irrelevantFilePath);
  stream.close();
  // Irrelevant directory, should not be reclaimed
  Path irrelevantDirPath = new Path(testDoneDirPath, "irrelevant");
  fs.mkdirs(irrelevantDirPath);
  Path doneAppHomeDir = new Path(new Path(new Path(testDoneDirPath,
      Long.toString(mainTestAppId.getClusterTimestamp())), "0000"), "001");
  // First application, untouched after creation
  Path appDirClean = new Path(doneAppHomeDir, appDirName);
  Path attemptDirClean = new Path(appDirClean, attemptDirName);
  fs.mkdirs(attemptDirClean);
  Path filePath = new Path(attemptDirClean, "test.log");
  stream = fs.create(filePath);
  stream.close();
  // Second application, one file touched after creation
  Path appDirHoldByFile = new Path(doneAppHomeDir, appDirName + "1");
  Path attemptDirHoldByFile
      = new Path(appDirHoldByFile, attemptDirName);
  fs.mkdirs(attemptDirHoldByFile);
  Path filePathHold = new Path(attemptDirHoldByFile, "test1.log");
  stream = fs.create(filePathHold);
  stream.close();
  // Third application, one dir touched after creation
  Path appDirHoldByDir = new Path(doneAppHomeDir, appDirName + "2");
  Path attemptDirHoldByDir = new Path(appDirHoldByDir, attemptDirName);
  fs.mkdirs(attemptDirHoldByDir);
  Path dirPathHold = new Path(attemptDirHoldByDir, "hold");
  fs.mkdirs(dirPathHold);
  // Fourth application, empty dirs
  Path appDirEmpty = new Path(doneAppHomeDir, appDirName + "3");
  Path attemptDirEmpty = new Path(appDirEmpty, attemptDirName);
  fs.mkdirs(attemptDirEmpty);
  Path dirPathEmpty = new Path(attemptDirEmpty, "empty");
  fs.mkdirs(dirPathEmpty);
  // Should retain all logs after this run
  MutableCounterLong dirsCleaned = store.metrics.getLogsDirsCleaned();
  long before = dirsCleaned.value();
  store.cleanLogs(testDoneDirPath, 10000);
  assertTrue(fs.exists(irrelevantDirPath));
  assertTrue(fs.exists(irrelevantFilePath));
  assertTrue(fs.exists(filePath));
  assertTrue(fs.exists(filePathHold));
  assertTrue(fs.exists(dirPathHold));
  assertTrue(fs.exists(dirPathEmpty));
  // Make sure the created dir is old enough
  Thread.sleep(2000);
  // Touch the second application
  stream = fs.append(filePathHold);
  stream.writeBytes("append");
  stream.close();
  // Touch the third application by creating a new dir
  fs.mkdirs(new Path(dirPathHold, "holdByMe"));
  store.cleanLogs(testDoneDirPath, 1000);
  // Verification after the second cleaner call
  assertTrue(fs.exists(irrelevantDirPath));
  assertTrue(fs.exists(irrelevantFilePath));
  assertTrue(fs.exists(filePathHold));
  assertTrue(fs.exists(dirPathHold));
  assertTrue(fs.exists(doneAppHomeDir));
  // appDirClean and appDirEmpty should be cleaned up
  assertFalse(fs.exists(appDirClean));
  assertFalse(fs.exists(appDirEmpty));
  assertEquals(before + 2L, dirsCleaned.value());
}