diff --git a/CHANGES.txt b/CHANGES.txt
index d99215c..47b4980 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -663,6 +663,9 @@
     MAPREDUCE-2428. start-mapred.sh script fails if HADOOP_HOME is not set.
     (tomwhite via eli)
 
+    MAPREDUCE-2445. Fix TestMiniMRWithDFSWithDistinctUsers to be a valid test.
+    (todd)
+
 Release 0.21.1 - Unreleased
 
   NEW FEATURES
diff --git a/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFSWithDistinctUsers.java b/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFSWithDistinctUsers.java
index a0551cd..a57e353 100644
--- a/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFSWithDistinctUsers.java
+++ b/src/test/mapred/org/apache/hadoop/mapred/TestMiniMRWithDFSWithDistinctUsers.java
@@ -24,16 +24,11 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.mapred.lib.IdentityMapper;
-import org.apache.hadoop.mapred.lib.IdentityReducer;
-import org.apache.hadoop.mapreduce.JobSubmissionFiles;
-import org.apache.hadoop.mapreduce.protocol.ClientProtocol;
-import org.apache.hadoop.mapreduce.split.JobSplitWriter;
-import org.apache.hadoop.mapreduce.split.JobSplit.SplitMetaInfo;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;
 import org.apache.hadoop.security.*;
 
 /**
@@ -41,8 +36,14 @@
  */
 public class TestMiniMRWithDFSWithDistinctUsers extends TestCase {
   static final UserGroupInformation DFS_UGI = createUGI("dfs", true); 
-  static final UserGroupInformation PI_UGI = createUGI("pi", false); 
-  static final UserGroupInformation WC_UGI = createUGI("wc", false); 
+  static final UserGroupInformation ALICE_UGI = createUGI("alice", false); 
+  static final UserGroupInformation BOB_UGI = createUGI("bob", false); 
+
+  MiniMRCluster mr = null;
+  MiniDFSCluster dfs = null;
+  FileSystem fs = null;
+  Configuration conf = new Configuration();
+  String jobTrackerName;
 
   static UserGroupInformation createUGI(String name, boolean issuper) {
     String group = issuper? "supergroup": name;
@@ -50,106 +51,73 @@
     return UserGroupInformation.createUserForTesting(name, new String[]{group});
   }
   
-  static void mkdir(FileSystem fs, String dir) throws IOException {
+  static void mkdir(FileSystem fs, String dir,
+                    String user, String group, short mode) throws IOException {
     Path p = new Path(dir);
     fs.mkdirs(p);
-    fs.setPermission(p, new FsPermission((short)0777));
+    fs.setPermission(p, new FsPermission(mode));
+    fs.setOwner(p, user, group);
   }
 
   // runs a sample job as a user (ugi)
-  RunningJob runJobAsUser(final JobConf job, UserGroupInformation ugi) 
+  void runJobAsUser(final JobConf job, UserGroupInformation ugi) 
   throws Exception {
-    ClientProtocol jobSubmitClient = 
-      TestSubmitJob.getJobSubmitClient(job, ugi);
-    org.apache.hadoop.mapreduce.JobID id = jobSubmitClient.getNewJobID();
-    
-    InputSplit[] splits = computeJobSplit(JobID.downgrade(id), job);
-    final Path jobSubmitDir = new Path(id.toString());
-    FileSystem fs = ugi.doAs(new PrivilegedExceptionAction<FileSystem>() {
-      public FileSystem run() throws IOException {
-        return jobSubmitDir.getFileSystem(job);
-      }
-    });
-    Path qJobSubmitDir = jobSubmitDir.makeQualified(fs);
-    uploadJobFiles(JobID.downgrade(id), splits, qJobSubmitDir, ugi, job);
-    
-    jobSubmitClient.submitJob(id, qJobSubmitDir.toString(), null);
-    
-    JobClient jc = new JobClient(job);
-    return jc.getJob(JobID.downgrade(id));
-  }
-  
-  // a helper api for split computation
-  private InputSplit[] computeJobSplit(JobID id, JobConf conf) 
-  throws IOException {
-    InputSplit[] splits = 
-      conf.getInputFormat().getSplits(conf, conf.getNumMapTasks());
-    conf.setNumMapTasks(splits.length);
-    return splits;
+    RunningJob rj = ugi.doAs(new PrivilegedExceptionAction<RunningJob>() {
+        public RunningJob run() throws IOException {
+          return JobClient.runJob(job);
+        }
+      });
+
+    rj.waitForCompletion();
+    assertEquals("SUCCEEDED", JobStatus.getJobRunState(rj.getJobState()));
   }
 
+  public void setUp() throws Exception {
+    dfs = new MiniDFSCluster(conf, 4, true, null);
 
-  // a helper api for split submission
-  private void uploadJobFiles(JobID id, InputSplit[] splits,
-                             Path jobSubmitDir, UserGroupInformation ugi, 
-                             final JobConf conf) 
-  throws Exception {
-    final Path confLocation = JobSubmissionFiles.getJobConfPath(jobSubmitDir);
-    FileSystem fs = ugi.doAs(new PrivilegedExceptionAction<FileSystem>() {
-      public FileSystem run() throws IOException {
-        return confLocation.getFileSystem(conf);
-      }
-    });
-    JobSplitWriter.createSplitFiles(jobSubmitDir, conf, fs, splits);
-    FsPermission perm = new FsPermission((short)0700);
-    
-    // localize conf
-    DataOutputStream confOut = FileSystem.create(fs, confLocation, perm);
-    conf.writeXml(confOut);
-    confOut.close();
-  }
-  
-  public void testDistinctUsers() throws Exception {
-    MiniMRCluster mr = null;
-    Configuration conf = new Configuration();
-    final MiniDFSCluster dfs = new MiniDFSCluster(conf, 4, true, null);
-    try {
-          
-      FileSystem fs = DFS_UGI.doAs(new PrivilegedExceptionAction<FileSystem>() {
+    fs = DFS_UGI.doAs(new PrivilegedExceptionAction<FileSystem>() {
         public FileSystem run() throws IOException {
           return dfs.getFileSystem();
         }
       });
-      mkdir(fs, "/user");
-      mkdir(fs, "/mapred");
+    // Home directories for users
+    mkdir(fs, "/user", "nobody", "nogroup", (short)01777);
+    mkdir(fs, "/user/alice", "alice", "nogroup", (short)0755);
+    mkdir(fs, "/user/bob", "bob", "nogroup", (short)0755);
 
-      UserGroupInformation MR_UGI = UserGroupInformation.getLoginUser(); 
-      mr = new MiniMRCluster(0, 0, 4, dfs.getFileSystem().getUri().toString(),
-           1, null, null, MR_UGI);
-      String jobTrackerName = "localhost:" + mr.getJobTrackerPort();
+    // staging directory root with sticky bit
+    UserGroupInformation MR_UGI = UserGroupInformation.getLoginUser(); 
+    mkdir(fs, "/staging", MR_UGI.getShortUserName(), "nogroup", (short)01777);
 
-      JobConf job1 = mr.createJobConf();
-      String input = "The quick brown fox\nhas many silly\n" 
-                     + "red fox sox\n";
-      Path inDir = new Path("/testing/distinct/input");
-      Path outDir = new Path("/testing/distinct/output");
-      TestMiniMRClasspath.configureWordCount(fs, jobTrackerName, job1, 
-                                             input, 2, 1, inDir, outDir);
-      runJobAsUser(job1, PI_UGI);
+    JobConf mrConf = new JobConf();
+    mrConf.set(JTConfig.JT_STAGING_AREA_ROOT, "/staging");
 
-      JobConf job2 = mr.createJobConf();
-      Path inDir2 = new Path("/testing/distinct/input2");
-      Path outDir2 = new Path("/testing/distinct/output2");
-      TestMiniMRClasspath.configureWordCount(fs, jobTrackerName, job2, 
-                                             input, 2, 1, inDir2, outDir2);
-      runJobAsUser(job2, WC_UGI);
-    } finally {
-      if (dfs != null) { dfs.shutdown(); }
-      if (mr != null) { mr.shutdown();}
-    }
+    mr = new MiniMRCluster(0, 0, 4, dfs.getFileSystem().getUri().toString(),
+                           1, null, null, MR_UGI, mrConf);
+    jobTrackerName = "localhost:" + mr.getJobTrackerPort();
+  }
+
+  public void tearDown() throws Exception {
+    if (mr != null) { mr.shutdown();}
+    if (dfs != null) { dfs.shutdown(); }
   }
   
-  public void testRestartWithDistinctUsers() {
-    
+  public void testDistinctUsers() throws Exception {
+    JobConf job1 = mr.createJobConf();
+    String input = "The quick brown fox\nhas many silly\n" 
+      + "red fox sox\n";
+    Path inDir = new Path("/testing/distinct/input");
+    Path outDir = new Path("/user/alice/output");
+    TestMiniMRClasspath.configureWordCount(fs, jobTrackerName, job1, 
+                                           input, 2, 1, inDir, outDir);
+    runJobAsUser(job1, ALICE_UGI);
+
+    JobConf job2 = mr.createJobConf();
+    Path inDir2 = new Path("/testing/distinct/input2");
+    Path outDir2 = new Path("/user/bob/output2");
+    TestMiniMRClasspath.configureWordCount(fs, jobTrackerName, job2, 
+                                           input, 2, 1, inDir2, outDir2);
+    runJobAsUser(job2, BOB_UGI);
   }
+
 }
diff --git a/src/test/mapred/org/apache/hadoop/mapred/TestSeveral.java b/src/test/mapred/org/apache/hadoop/mapred/TestSeveral.java
index 1310679..d5e063f 100644
--- a/src/test/mapred/org/apache/hadoop/mapred/TestSeveral.java
+++ b/src/test/mapred/org/apache/hadoop/mapred/TestSeveral.java
@@ -93,10 +93,10 @@
           }
         });
 
-        TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/user");
-        TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/mapred");
-        TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, 
-            conf.get(JTConfig.JT_STAGING_AREA_ROOT));
+        TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/user", "mapred", "mapred", (short)01777);
+        TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/mapred", "mapred", "mapred", (short)01777);
+        TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, conf.get(JTConfig.JT_STAGING_AREA_ROOT),
+                                                 "mapred", "mapred", (short)01777);
 
         UserGroupInformation MR_UGI = UserGroupInformation.getLoginUser(); 
 
diff --git a/src/test/mapred/org/apache/hadoop/mapred/TestSubmitJob.java b/src/test/mapred/org/apache/hadoop/mapred/TestSubmitJob.java
index 39a92ce..ffb73e9 100644
--- a/src/test/mapred/org/apache/hadoop/mapred/TestSubmitJob.java
+++ b/src/test/mapred/org/apache/hadoop/mapred/TestSubmitJob.java
@@ -206,10 +206,11 @@
             return dfs.getFileSystem();
           }
         });
-      TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/user");
-      TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/mapred");
-      TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, 
-          conf.get(JTConfig.JT_STAGING_AREA_ROOT));
+      TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/user", "mapred", "mapred", (short)01777);
+      TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, "/mapred", "mapred", "mapred", (short)01777);
+      TestMiniMRWithDFSWithDistinctUsers.mkdir(fs, conf.get(JTConfig.JT_STAGING_AREA_ROOT),
+                                               "mapred", "mapred", (short)01777);
+
       UserGroupInformation MR_UGI = UserGroupInformation.getLoginUser();
       mr = new MiniMRCluster(0, 0, 1, dfs.getFileSystem().getUri().toString(),
                              1, null, null, MR_UGI);