conf2.setLong("mapreduce.tasktracker.local.cache.numberdirectories", 3);
//The goal is to get down to 15.75K and 2 dirs
conf2.setFloat("mapreduce.tasktracker.cache.local.keep.pct", 0.75f);
conf2.setLong("mapreduce.tasktracker.distributedcache.checkperiod", CACHE_DELETE_PERIOD_MS);
refreshConf(conf2);
TrackerDistributedCacheManager manager =
new TrackerDistributedCacheManager(conf2, taskController);
try {
manager.startCleanupThread();
FileSystem localfs = FileSystem.getLocal(conf2);
String userName = getJobOwnerName();
conf2.set("user.name", userName);
//Here we are testing the LRU. In this case we will add in 4 cache entries
// 2 of them are 8k each and 2 of them are very small. We want to verify
// That they are deleted in LRU order.
// So what we will do is add in the two large files first, 1 then 2, and
// then one of the small ones 3. We will then release them in opposite
// order 3, 2, 1.
//
// Finally we will add in the last small file. This last file should push
// us over the 3 entry limit to trigger a cleanup. So LRU order is 3, 2, 1
// And we will only delete 2 entries so that should leave 1 un touched
// but 3 and 2 deleted
Path thirdCacheFile = new Path(TEST_ROOT_DIR, "thirdcachefile");
Path fourthCacheFile = new Path(TEST_ROOT_DIR, "fourthcachefile");
// Adding two more small files, so it triggers the number of sub directory
// limit but does not trigger the file size limit.
createTempFile(thirdCacheFile, 1);
createTempFile(fourthCacheFile, 1);
FileStatus stat = fs.getFileStatus(firstCacheFilePublic);
CacheFile cfile1 = new CacheFile(firstCacheFilePublic.toUri(),
CacheFile.FileType.REGULAR, true,
stat.getModificationTime(),
true);
Path firstLocalCache = manager.getLocalCache(firstCacheFilePublic.toUri(), conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(firstCacheFilePublic), false,
fs.getFileStatus(firstCacheFilePublic).getModificationTime(), true,
cfile1);
stat = fs.getFileStatus(secondCacheFilePublic);
CacheFile cfile2 = new CacheFile(secondCacheFilePublic.toUri(),
CacheFile.FileType.REGULAR, true,
stat.getModificationTime(),
true);
Path secondLocalCache = manager.getLocalCache(secondCacheFilePublic.toUri(), conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(secondCacheFilePublic), false,
fs.getFileStatus(secondCacheFilePublic).getModificationTime(), true,
cfile2);
stat = fs.getFileStatus(thirdCacheFile);
CacheFile cfile3 = new CacheFile(thirdCacheFile.toUri(),
CacheFile.FileType.REGULAR, true,
stat.getModificationTime(),
true);
Path thirdLocalCache = manager.getLocalCache(thirdCacheFile.toUri(), conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(thirdCacheFile), false,
fs.getFileStatus(thirdCacheFile).getModificationTime(), true,
cfile3);
manager.releaseCache(cfile3.getStatus());
manager.releaseCache(cfile2.getStatus());
manager.releaseCache(cfile1.getStatus());
// Getting the fourth cache will make the number of sub directories becomes
// 4 which is greater than 3. So the released cache will be deleted.
stat = fs.getFileStatus(fourthCacheFile);
CacheFile cfile4 = new CacheFile(fourthCacheFile.toUri(),
CacheFile.FileType.REGULAR, true,
stat.getModificationTime(),
true);
Path fourthLocalCache = manager.getLocalCache(fourthCacheFile.toUri(), conf2,
TaskTracker.getPrivateDistributedCacheDir(userName),
fs.getFileStatus(fourthCacheFile), false,
fs.getFileStatus(fourthCacheFile).getModificationTime(), true,
cfile4);
checkCacheDeletion(localfs, secondLocalCache, "DistributedCache failed " +
"deleting second cache LRU order");
checkCacheDeletion(localfs, thirdLocalCache,
"DistributedCache failed deleting third" +
" cache LRU order.");
checkCacheNOTDeletion(localfs, firstLocalCache, "DistributedCache failed " +
"Deleted first cache LRU order.");
checkCacheNOTDeletion(localfs, fourthCacheFile, "DistributedCache failed " +
"Deleted fourth cache LRU order.");
// Clean up the files created in this test
new File(thirdCacheFile.toString()).delete();
new File(fourthCacheFile.toString()).delete();
} finally {
manager.stopCleanupThread();
}
}