MAPREDUCE-2928. MR-2413 improvements. Contributed by Eli Collins.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-205@1174546 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -205,6 +205,8 @@ Release 0.20.205.0 - 2011.09.12
 
   IMPROVEMENTS
 
+    MAPREDUCE-2928. MR-2413 improvements (Eli Collins via mattf)
+
     HADOOP-7655. provide a small validation script that smoke tests the installed
     cluster. (Arpit Gupta via mattf)
 

src/mapred/org/apache/hadoop/mapred/DefaultTaskController.java

@@ -69,7 +69,7 @@ public class DefaultTaskController extends TaskController {
   public void createLogDir(TaskAttemptID taskID, 
 		  			boolean isCleanup) throws IOException {
     TaskLog.createTaskAttemptLogDir(taskID, isCleanup, 
-                                    localStorage.getGoodLocalDirs());
+                                    localStorage.getDirs());
   }
   
   /**
@@ -243,7 +243,7 @@ public class DefaultTaskController extends TaskController {
     FileUtil.fullyDelete( subDirPath );
     
     //Delete the subDir in all good <mapred.local.dirs>/userlogs 
-    String [] localDirs = localStorage.getGoodLocalDirs();
+    String [] localDirs = localStorage.getDirs();
     for(String localdir : localDirs) {
     	String dirPath = localdir + File.separatorChar + 
     					TaskLog.USERLOGS_DIR_NAME + File.separatorChar +
@@ -271,9 +271,9 @@ public class DefaultTaskController extends TaskController {
   }
 
   @Override
-  public void setup(LocalDirAllocator allocator, LocalStorage l) {
+  public void setup(LocalDirAllocator allocator, LocalStorage localStorage) {
     this.allocator = allocator;
-    this.localStorage = l;
+    this.localStorage = localStorage;
   }
   
 }

src/mapred/org/apache/hadoop/mapred/LinuxTaskController.java

@@ -156,7 +156,7 @@ class LinuxTaskController extends TaskController {
     List<String> command = new ArrayList<String>(
       Arrays.asList(taskControllerExe, 
                     user,
-                    localStorage.getGoodLocalDirsString(),
+                    localStorage.getDirsString(),
                     Integer.toString(Commands.INITIALIZE_JOB.getValue()),
                     jobid,
                     credentials.toUri().getPath().toString(),
@@ -222,7 +222,7 @@ class LinuxTaskController extends TaskController {
       String[] command = 
         new String[]{taskControllerExe, 
           user,
-          localStorage.getGoodLocalDirsString(),
+          localStorage.getDirsString(),
           Integer.toString(Commands.LAUNCH_TASK_JVM.getValue()),
           jobId,
           attemptId,
@@ -263,7 +263,7 @@ class LinuxTaskController extends TaskController {
     String[] command = 
       new String[]{taskControllerExe, 
                    user,
-                   localStorage.getGoodLocalDirsString(),
+                   localStorage.getDirsString(),
                    Integer.toString(Commands.DELETE_AS_USER.getValue()),
                    subDir};
     ShellCommandExecutor shExec = new ShellCommandExecutor(command);
@@ -278,7 +278,7 @@ class LinuxTaskController extends TaskController {
     String[] command = 
       new String[]{taskControllerExe, 
                    user,
-                   localStorage.getGoodLocalDirsString(),
+                   localStorage.getDirsString(),
                    Integer.toString(Commands.DELETE_LOG_AS_USER.getValue()),
                    subDir};
     ShellCommandExecutor shExec = new ShellCommandExecutor(command);
@@ -294,7 +294,7 @@ class LinuxTaskController extends TaskController {
     String[] command = 
       new String[]{taskControllerExe, 
                    user,
-                   localStorage.getGoodLocalDirsString(),
+                   localStorage.getDirsString(),
                    Integer.toString(Commands.SIGNAL_TASK.getValue()),
                    Integer.toString(taskPid),
                    Integer.toString(signal.getValue())};
@@ -362,7 +362,7 @@ class LinuxTaskController extends TaskController {
     String[] taskControllerCmd = new String[4 + command.size()];
     taskControllerCmd[0] = taskControllerExe;
     taskControllerCmd[1] = user;
-    taskControllerCmd[2] = localStorage.getGoodLocalDirsString();
+    taskControllerCmd[2] = localStorage.getDirsString();
     taskControllerCmd[3] = Integer.toString(
         Commands.RUN_COMMAND_AS_USER.getValue());
     int i = 4;

src/mapred/org/apache/hadoop/mapred/TaskController.java

@@ -58,10 +58,6 @@ public abstract class TaskController implements Configurable {
   protected static final String COMMAND_FILE = "taskjvm.sh";
   
   protected LocalDirAllocator allocator;
-
-  /**
-   * LocalStorage of TaskTracker
-   */
   protected LocalStorage localStorage;
 
   final public static FsPermission TASK_LAUNCH_SCRIPT_PERMISSION =
@@ -71,8 +67,8 @@ public abstract class TaskController implements Configurable {
     return conf;
   }
 
-  public LocalStorage getLocalStorage() {
-	  return localStorage;
+  public String[] getLocalDirs() {
+    return localStorage.getDirs();
   }
   
   public void setConf(Configuration conf) {
@@ -82,10 +78,10 @@ public abstract class TaskController implements Configurable {
   /**
    * Does initialization and setup.
    * @param allocator the local dir allocator to use
-   * @param l TaskTracker's LocalStorage object
+   * @param localStorage local storage to obtain dirs from
    */
   public abstract void setup(LocalDirAllocator allocator,
-      LocalStorage l) throws IOException;
+      LocalStorage localStorage) throws IOException;
   
   /**
    * Create all of the directories necessary for the job to start and download

src/mapred/org/apache/hadoop/mapred/TaskTracker.java

@@ -20,7 +20,6 @@ package org.apache.hadoop.mapred;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.RandomAccessFile;
@@ -62,7 +61,6 @@ import org.apache.hadoop.filecache.TrackerDistributedCacheManager;
 import org.apache.hadoop.mapreduce.server.tasktracker.*;
 import org.apache.hadoop.mapreduce.server.tasktracker.userlogs.*;
 import org.apache.hadoop.fs.DF;
-import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
@@ -76,15 +74,12 @@ import org.apache.hadoop.io.SecureIOUtils;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.ipc.Server;
-import org.apache.hadoop.mapred.QueueManager.QueueACL;
 import org.apache.hadoop.mapred.CleanupQueue.PathDeletionContext;
-import org.apache.hadoop.mapred.TaskController.DeletionContext;
 import org.apache.hadoop.mapred.TaskLog.LogFileDetail;
 import org.apache.hadoop.mapred.TaskLog.LogName;
 import org.apache.hadoop.mapred.TaskStatus.Phase;
 import org.apache.hadoop.mapred.TaskTrackerStatus.TaskTrackerHealthStatus;
 import org.apache.hadoop.mapred.pipes.Submitter;
-import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.mapreduce.security.SecureShuffleUtils;
 import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
@@ -101,7 +96,6 @@ import org.apache.hadoop.util.ProcfsBasedProcessTree;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.hadoop.util.RunJar;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.VersionInfo;
 import org.apache.hadoop.util.DiskChecker.DiskErrorException;
@@ -167,94 +161,71 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
   volatile boolean running = true;
 
   /**
-   *  TaskTracker internal class only.
-   *  Manages the lists of good mapred local dirs and bad mapred local dirs.
+   * Manages TT local storage directories.
    */
-  public static class LocalStorage {
-    private List<String> goodLocalDirs = new ArrayList<String>();
-    private List<String> badLocalDirs = new ArrayList<String>();
-    private boolean diskFailed = false;
+  static class LocalStorage {
+    private List<String> localDirs;
+    private int numFailures;
 
-    /**
-     * TaskTracker internal only
-     */
-    public LocalStorage(String[] localDirs) {
-      for (String s : localDirs) {
-        goodLocalDirs.add(s);
-      }
+    public LocalStorage(String[] dirs) {
+      localDirs = new ArrayList<String>();
+      localDirs.addAll(Arrays.asList(dirs));
     }
 
     /**
-     * @return good mapred local dirs list
+     * @return the current valid directories 
      */
-    synchronized String[] getGoodLocalDirs() {
-      String[] rv = new String[goodLocalDirs.size()];
-      return goodLocalDirs.toArray(rv);
+    synchronized String[] getDirs() {
+      return localDirs.toArray(new String[localDirs.size()]);
     }
 
     /**
-     * @return good mapred local dirs list as a commma seperated string
+     * @return the current valid dirs as comma separated string
      */
-    synchronized String getGoodLocalDirsString() {
-      StringBuffer sb = new StringBuffer();
-      for (String s : goodLocalDirs) {
-        if (sb.length() > 0) {
-          sb.append(",");
-        }
-        sb.append(s);
-      }
-      return sb.toString();
+    synchronized String getDirsString() {
+      return StringUtils.join(",", localDirs);
     }
 
     /**
-     * @return bad mapred local dirs list
+     * @return the number of valid local directories
      */
-    synchronized String[] getBadLocalDirs() {
-      String[] rv = new String[badLocalDirs.size()];
-      return badLocalDirs.toArray(rv);
+    synchronized int numDirs() {
+      return localDirs.size();
     }
 
     /**
-     * @return true if a disk has failed since the last
-     * time this method was called
+     * @return the number of directory failures
      */
-    synchronized boolean isDiskFailed() {
-      boolean rv = diskFailed;
-      diskFailed = false;
-      return rv;
+    synchronized int numFailures() {
+      return numFailures;
     }
 
     /**
-     * Check if the given local directories
-     * (and parent directories, if necessary) can be created.
-     * Updates the list of good mapred local dirs and the list of bad local
-     * dirs.
-     * @throws DiskErrorException if all local directories are not writable
+     * Check the current set of local directories, updating the list
+     * of valid directories if necessary.
+     * @throws DiskErrorException if no directories are writable
      */
-    synchronized void checkLocalDirs()
-        throws DiskErrorException {
-      for (String s : getGoodLocalDirs()) {
+    synchronized void checkDirs() throws DiskErrorException {
+      for (String dir : localDirs) {
         try {
-          DiskChecker.checkDir(new File(s));
-        } catch(DiskErrorException e) {
-          LOG.warn("Task Tracker localdir error " + e.getMessage()
-            + ", removing from good locadirs");
-          goodLocalDirs.remove(s);
-          badLocalDirs.add(s);
-          diskFailed = true;
+          DiskChecker.checkDir(new File(dir));
+        } catch (DiskErrorException de) {
+          LOG.warn("TaskTracker local dir " + dir + " error " + 
+              de.getMessage() + ", removing from local dirs");
+          localDirs.remove(dir);
+          numFailures++;
         }
       }
-
-      // no good local dirs ?
-      if(goodLocalDirs.size() < 1) {
+      if (localDirs.isEmpty()) {
         throw new DiskErrorException(
-            "All mapred local directories are not writable.");
+            "No mapred local directories are writable");
       }
     }
   }
 
   private LocalStorage localStorage;
   private long lastCheckDirsTime;
+  private int lastNumFailures;
   private LocalDirAllocator localDirAllocator;
   String taskTrackerName;
   String localHostname;
@@ -680,7 +651,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
    * @throws IOException
    */
   private void deleteUserDirectories(Configuration conf) throws IOException {
-    for(String root: localStorage.getGoodLocalDirs()) {
+    for(String root: localStorage.getDirs()) {
       for(FileStatus status: localFs.listStatus(new Path(root, SUBDIR))) {
         String owner = status.getOwner();
         String path = status.getPath().getName();
@@ -716,11 +687,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
        fConf.get("mapred.tasktracker.dns.nameserver","default"));
     }
 
-    localStorage.checkLocalDirs();
-    if (localStorage.isDiskFailed()) {
-      // Ignore current disk failures. They are being handled now.
-    }
-    String dirs = localStorage.getGoodLocalDirsString();
+    final String dirs = localStorage.getDirsString();
     fConf.setStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY, dirs);
     LOG.info("Good mapred local directories are: " + dirs);
     taskController.setConf(fConf);
@@ -730,23 +697,25 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
 
     deleteUserDirectories(fConf);
 
+    // NB: deleteLocalFiles uses the configured local dirs, but does not 
+    // fail if a local directory has failed. 
     fConf.deleteLocalFiles(SUBDIR);
     final FsPermission ttdir = FsPermission.createImmutable((short) 0755);
-    for (String s : localStorage.getGoodLocalDirs()) {
+    for (String s : localStorage.getDirs()) {
       localFs.mkdirs(new Path(s, SUBDIR), ttdir);
     }
     fConf.deleteLocalFiles(TT_PRIVATE_DIR);
     final FsPermission priv = FsPermission.createImmutable((short) 0700);
-    for (String s : localStorage.getGoodLocalDirs()) {
+    for (String s : localStorage.getDirs()) {
       localFs.mkdirs(new Path(s, TT_PRIVATE_DIR), priv);
     }
     fConf.deleteLocalFiles(TT_LOG_TMP_DIR);
     final FsPermission pub = FsPermission.createImmutable((short) 0755);
-    for (String s : localStorage.getGoodLocalDirs()) {
+    for (String s : localStorage.getDirs()) {
       localFs.mkdirs(new Path(s, TT_LOG_TMP_DIR), pub);
     }
     // Create userlogs directory under all good mapred-local-dirs
-    for (String s : localStorage.getGoodLocalDirs()) {
+    for (String s : localStorage.getDirs()) {
       Path userLogsDir = new Path(s, TaskLog.USERLOGS_DIR_NAME);
       if (!localFs.exists(userLogsDir)) {
         localFs.mkdirs(userLogsDir, pub);
@@ -845,7 +814,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
     reduceLauncher.start();
 
     // create a localizer instance
-    setLocalizer(new Localizer(localFs, localStorage.getGoodLocalDirs()));
+    setLocalizer(new Localizer(localFs, localStorage.getDirs()));
 
     //Start up node health checker service.
     if (shouldStartHealthMonitor(this.fConf)) {
@@ -856,13 +825,6 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
       fConf.getBoolean(TT_OUTOFBAND_HEARBEAT, false);
   }
 
-  /**
-   * @return TaskTracker's LocalStorage object
-   */
-  LocalStorage getLocalStorage() {
-    return localStorage;
-  }
-
   private void createInstrumentation() {
     Class<? extends TaskTrackerInstrumentation> metricsInst =
         getInstrumentationClass(fConf);
@@ -1316,7 +1278,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
                                 RunningJob rjob) throws IOException {
     synchronized (tip) {
       jobConf.set(JobConf.MAPRED_LOCAL_DIR_PROPERTY,
-          localStorage.getGoodLocalDirsString());
+                  localStorage.getDirsString());
       tip.setJobConf(jobConf);
       tip.setUGI(rjob.ugi);
       tip.launchTask(rjob);
@@ -1406,14 +1368,6 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
     fConf = conf;
   }
 
-  void setLocalStorage(LocalStorage in) {
-	localStorage = in;  
-  }
-	  
-  void setLocalDirAllocator(LocalDirAllocator in) {
-	localDirAllocator = in;  
-  }
-  
   /**
    * Start with the local machine name, and the default JobTracker
    */
@@ -1451,10 +1405,11 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
 
     fConf = new JobConf(conf);
     localStorage = new LocalStorage(fConf.getLocalDirs());
-    localStorage.checkLocalDirs();
+    localStorage.checkDirs();
     taskController = 
       (TaskController) ReflectionUtils.newInstance(taskControllerClass, fConf);
     taskController.setup(localDirAllocator, localStorage);
+    lastNumFailures = localStorage.numFailures();
 
     // create user log manager
     setUserLogManager(new UserLogManager(conf, taskController));
@@ -1605,10 +1560,12 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
 
         now = System.currentTimeMillis();
         if (now > (lastCheckDirsTime + diskHealthCheckInterval)) {
-          localStorage.checkLocalDirs();
+          localStorage.checkDirs();
           lastCheckDirsTime = now;
-          // If any of the good disks failed, re-init the task tracker
-          if (localStorage.isDiskFailed()) {
+          int numFailures = localStorage.numFailures();
+          // Re-init the task tracker if there were any new failures
+          if (numFailures > lastNumFailures) {
+            lastNumFailures = numFailures;
             return State.STALE;
           }
         }
@@ -1994,7 +1951,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
                                          jobDir.substring(userDir.length()));
     directoryCleanupThread.addToQueue(jobCleanup);
     
-    for (String str : localStorage.getGoodLocalDirs()) {
+    for (String str : localStorage.getDirs()) {
       Path ttPrivateJobDir = FileSystem.getLocal(fConf).makeQualified(
         new Path(str, TaskTracker.getPrivateDirForJob(user, jobId.toString())));
       PathDeletionContext ttPrivateJobCleanup =
@@ -2116,7 +2073,7 @@ public class TaskTracker implements MRConstants, TaskUmbilicalProtocol,
   
   private long getFreeSpace() throws IOException {
     long biggestSeenSoFar = 0;
-    String[] localDirs = localStorage.getGoodLocalDirs();
+    String[] localDirs = localStorage.getDirs();
     for (int i = 0; i < localDirs.length; i++) {
       DF df = null;
       if (localDirsDf.containsKey(localDirs[i])) {

src/mapred/org/apache/hadoop/mapred/UserLogCleaner.java

@@ -233,7 +233,7 @@ public class UserLogCleaner extends Thread {
 	//If <hadoop.log.dir>/userlogs/jobid not found, then get user from 
 	//any one of existing <mapred.local.dir>/userlogs/jobid path(s)
 	String[] localDirs = 
-	   userLogManager.getTaskController().getLocalStorage().getGoodLocalDirs();
+	   userLogManager.getTaskController().getLocalDirs();
 	for(String localDir : localDirs) {
 		try{
 		   logRoot = localDir + File.separator + TaskLog.USERLOGS_DIR_NAME;

src/test/org/apache/hadoop/filecache/TestTrackerDistributedCacheManager.java

@@ -42,7 +42,7 @@ import org.apache.hadoop.mapred.JobID;
 import org.apache.hadoop.mapred.JobLocalizer;
 import org.apache.hadoop.mapred.TaskController;
 import org.apache.hadoop.mapred.TaskTracker;
-import org.apache.hadoop.mapred.TaskTracker.LocalStorage;
+import org.apache.hadoop.mapred.UtilsForTests;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -114,8 +114,8 @@ public class TestTrackerDistributedCacheManager extends TestCase {
         taskControllerClass, conf);
 
     // setup permissions for mapred local dir
-    taskController.setup(localDirAllocator,
-        new LocalStorage(conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY)));
+    UtilsForTests.setupTC(taskController, localDirAllocator,
+        conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY));
 
     // Create the temporary cache files to be used in the tests.
     firstCacheFile = new Path(TEST_ROOT_DIR, "firstcachefile");
@@ -130,8 +130,8 @@ public class TestTrackerDistributedCacheManager extends TestCase {
   
   protected void refreshConf(Configuration conf) throws IOException {
     taskController.setConf(conf);
-    taskController.setup(localDirAllocator,
-        new LocalStorage(conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY)));
+    UtilsForTests.setupTC(taskController, localDirAllocator,
+        conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY));
   }
 
   /**

src/test/org/apache/hadoop/mapred/ClusterWithLinuxTaskController.java

@@ -33,7 +33,6 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.mapred.TaskTracker.LocalStorage;
@@ -76,7 +75,7 @@ public class ClusterWithLinuxTaskController extends TestCase {
         + "/task-controller";
 
     @Override
-    public void setup(LocalDirAllocator allocator, LocalStorage l)
+    public void setup(LocalDirAllocator allocator, LocalStorage localStorage)
         throws IOException {
       // get the current ugi and set the task controller group owner
       getConf().set(TT_GROUP, taskTrackerSpecialGroup);
@@ -84,7 +83,7 @@ public class ClusterWithLinuxTaskController extends TestCase {
       // write configuration file
       configurationFile = createTaskControllerConf(System
           .getProperty(TASKCONTROLLER_PATH), getConf());
-      super.setup(allocator, l);
+      super.setup(allocator, localStorage);
     }
 
     protected String getTaskControllerExecutablePath() {
@@ -211,11 +210,7 @@ public class ClusterWithLinuxTaskController extends TestCase {
     PrintWriter writer =
         new PrintWriter(new FileOutputStream(configurationFile));
 
-    //writer.println(String.format("mapred.local.dir=%s", conf.
-    //    get(JobConf.MAPRED_LOCAL_DIR_PROPERTY)));
-
-    writer
-        .println(String.format("hadoop.log.dir=%s", TaskLog.getBaseLogDir()));
+    writer.println(String.format("hadoop.log.dir=%s", TaskLog.getBaseLogDir()));
     writer.println(String.format(TT_GROUP + "=%s", conf.get(TT_GROUP)));
 
     writer.flush();

src/test/org/apache/hadoop/mapred/TestJvmManager.java

@@ -28,7 +28,6 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.filecache.TrackerDistributedCacheManager;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.LocalDirAllocator;
-import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.JvmManager.JvmManagerForType;
 import org.apache.hadoop.mapred.JvmManager.JvmManagerForType.JvmRunner;
 import org.apache.hadoop.mapred.TaskTracker.LocalStorage;

src/test/org/apache/hadoop/mapred/TestTaskTrackerLocalization.java

@@ -188,8 +188,7 @@ public class TestTaskTrackerLocalization extends TestCase {
     // setup task controller
     taskController = getTaskController();
     taskController.setConf(trackerFConf);
-    taskController.setup(lDirAlloc,
-                         new LocalStorage(trackerFConf.getLocalDirs()));
+    taskController.setup(lDirAlloc, new LocalStorage(trackerFConf.getLocalDirs()));
     tracker.setTaskController(taskController);
     tracker.setLocalizer(new Localizer(tracker.getLocalFileSystem(),localDirs));
   }

src/test/org/apache/hadoop/mapred/TestTrackerDistributedCacheManagerWithLinuxTaskController.java

@@ -66,9 +66,8 @@ public class TestTrackerDistributedCacheManagerWithLinuxTaskController extends
     String execPath = path + "/task-controller";
     ((MyLinuxTaskController)taskController).setTaskControllerExe(execPath);
     taskController.setConf(conf);
-    taskController.setup(
-        new LocalDirAllocator(JobConf.MAPRED_LOCAL_DIR_PROPERTY),
-        new LocalStorage(conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY)));
+    UtilsForTests.setupTC(taskController, localDirAllocator,
+        conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY));
   }
 
   @Override

src/test/org/apache/hadoop/mapred/TestUserLogCleanup.java

@@ -20,7 +20,6 @@ package org.apache.hadoop.mapred;
 import java.io.File;
 import java.io.IOException;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.LocalDirAllocator;
@@ -30,7 +29,6 @@ import org.apache.hadoop.mapreduce.JobContext;
 import org.apache.hadoop.mapreduce.server.tasktracker.Localizer;
 import org.apache.hadoop.mapreduce.server.tasktracker.userlogs.*;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.util.ReflectionUtils;
 
 import static org.junit.Assert.*;
 
@@ -92,16 +90,13 @@ public class TestUserLogCleanup {
     tt.setConf(new JobConf(conf));
     LocalDirAllocator localDirAllocator = 
     					new LocalDirAllocator("mapred.local.dir");
-    tt.setLocalDirAllocator(localDirAllocator);
-    LocalStorage localStorage = new LocalStorage(conf.getLocalDirs());
-    localStorage.checkLocalDirs();
-    tt.setLocalStorage(localStorage);
     localizer = new Localizer(FileSystem.get(conf), conf
         .getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY));
     tt.setLocalizer(localizer);
     userLogManager = new UtilsForTests.InLineUserLogManager(conf);
     TaskController taskController = userLogManager.getTaskController();
-    taskController.setup(localDirAllocator, localStorage);
+    taskController.setup(localDirAllocator,
+        new LocalStorage(conf.getStrings(JobConf.MAPRED_LOCAL_DIR_PROPERTY)));
     tt.setTaskController(taskController);
     userLogCleaner = userLogManager.getUserLogCleaner();
     userLogCleaner.setClock(myClock);

src/test/org/apache/hadoop/mapred/UtilsForTests.java

@@ -32,6 +32,7 @@ import java.util.Properties;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.logging.Log;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.LocalDirAllocator;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.permission.FsPermission;
@@ -49,6 +50,7 @@ import org.apache.hadoop.io.SequenceFile.CompressionType;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.CleanupQueue.PathDeletionContext;
 import org.apache.hadoop.mapred.SortValidator.RecordStatsChecker.NonSplitableSequenceFileInputFormat;
+import org.apache.hadoop.mapred.TaskTracker.LocalStorage;
 import org.apache.hadoop.mapred.lib.IdentityMapper;
 import org.apache.hadoop.mapred.lib.IdentityReducer;
 import org.apache.hadoop.mapreduce.server.tasktracker.userlogs.UserLogEvent;
@@ -878,5 +880,17 @@ public class UtilsForTests {
     return tmpTaskTracker;
   }
 
+  /**
+   * Setup the given task controller
+   * @param taskController to setup
+   * @param localAlloc local allocator to use
+   * @param dirs to use for local storage
+   * @throws IOException
+   */
+  public static void setupTC(TaskController taskController, 
+      LocalDirAllocator localAlloc, String[] dirs) throws IOException {
+    // Hide LocalStorage from callers
+    taskController.setup(localAlloc, new LocalStorage(dirs));
+  }
 }