HADOOP-1876. Persist statuses of completed jobs in HDFS so that the JobClient can query and get information about decommissioned jobs and also across JobTracker restarts. Contributed by Alejandro Abdelnur.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@611527 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -217,6 +217,16 @@ Trunk (unreleased changes)
     HADOOP-2464. Unit tests for chmod, chown, and chgrp using DFS.
     (Raghu Angadi)
 
+    HADOOP-1876. Persist statuses of completed jobs in HDFS so that the
+    JobClient can query and get information about decommissioned jobs and also
+    across JobTracker restarts.
+    Configuration changes to hadoop-default.xml:
+      add mapred.job.tracker.persist.jobstatus.active (default value of false)
+      add mapred.job.tracker.persist.jobstatus.hours (default value of 0)
+      add mapred.job.tracker.persist.jobstatus.dir (default value of
+                                                    /jobtracker/jobsInfo)
+    (Alejandro Abdelnur via acmurthy) 
+
   OPTIMIZATIONS
 
     HADOOP-1898.  Release the lock protecting the last time of the last stack

conf/hadoop-default.xml

@@ -893,6 +893,33 @@ creations/deletions), or "all".</description>
   </description>
 </property>
 
+  <property>
+    <name>mapred.job.tracker.persist.jobstatus.active</name>
+    <value>false</value>
+    <description>Indicates if persistency of job status information is
+      active or not.
+    </description>
+  </property>
+
+  <property>
+  <name>mapred.job.tracker.persist.jobstatus.hours</name>
+  <value>0</value>
+  <description>The number of hours job status information is persisted in DFS.
+    The job status information will be available after it drops of the memory
+    queue and between jobtracker restarts. With a zero value the job status
+    information is not persisted at all in DFS.
+  </description>
+</property>
+
+  <property>
+    <name>mapred.job.tracker.persist.jobstatus.dir</name>
+    <value>/jobtracker/jobsInfo</value>
+    <description>The directory where the job status information is persisted
+      in a file system to be available after it drops of the memory queue and
+      between jobtracker restarts.
+    </description>
+  </property>
+
 <!-- ipc properties -->
 
 <property>

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

@@ -0,0 +1,312 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */  
+package org.apache.hadoop.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+
+import java.io.IOException;
+
+/**
+ * Persists and retrieves the Job info of a job into/from DFS.
+ * <p/>
+ * If the retain time is zero jobs are not persisted.
+ * <p/>
+ * A daemon thread cleans up job info files older than the retain time
+ * <p/>
+ * The retain time can be set with the 'persist.jobstatus.hours'
+ * configuration variable (it is in hours).
+ */
+public class CompletedJobStatusStore implements Runnable {
+  private boolean active;
+  private String jobInfoDir;
+  private long retainTime;
+  private FileSystem fs;
+  private static final String JOB_INFO_STORE_DIR = "/jobtracker/jobsInfo";
+
+  public static final Log LOG =
+          LogFactory.getLog(CompletedJobStatusStore.class);
+
+  private static long HOUR = 1000 * 60 * 60;
+  private static long SLEEP_TIME = 1 * HOUR;
+
+  CompletedJobStatusStore(Configuration conf) throws IOException {
+    active =
+      conf.getBoolean("mapred.job.tracker.persist.jobstatus.active", false);
+
+    if (active) {
+      fs = FileSystem.get(conf);
+      retainTime =
+        conf.getInt("mapred.job.tracker.persist.jobstatus.hours", 0) * HOUR;
+
+      jobInfoDir =
+        conf.get("mapred.job.tracker.persist.jobstatus.dir", JOB_INFO_STORE_DIR);
+
+      Path path = new Path(jobInfoDir);
+      if (!fs.exists(path)) {
+        fs.mkdirs(path);
+      }
+
+      if (retainTime == 0) {
+        // as retain time is zero, all stored jobstatuses are deleted.
+        deleteJobStatusDirs();
+      }
+    }
+  }
+
+  /**
+   * Indicates if job status persistency is active or not.
+   *
+   * @return TRUE if active, FALSE otherwise.
+   */
+  public boolean isActive() {
+    return active;
+  }
+
+  public void run() {
+    if (retainTime > 0) {
+      while (true) {
+        deleteJobStatusDirs();
+        try {
+          Thread.sleep(SLEEP_TIME);
+        }
+        catch (InterruptedException ex) {
+          break;
+        }
+      }
+    }
+  }
+
+  private void deleteJobStatusDirs() {
+    try {
+      long currentTime = System.currentTimeMillis();
+      Path[] jobInfoFiles = fs.listPaths(
+              new Path[]{new Path(jobInfoDir)});
+
+      //noinspection ForLoopReplaceableByForEach
+      for (Path jobInfo : jobInfoFiles) {
+        try {
+          FileStatus status = fs.getFileStatus(jobInfo);
+          if ((currentTime - status.getModificationTime()) > retainTime) {
+            fs.delete(jobInfo);
+          }
+        }
+        catch (IOException ie) {
+          LOG.warn("Could not do housekeeping for [ " +
+                  jobInfo + "] job info : " + ie.getMessage(), ie);
+        }
+      }
+    }
+    catch (IOException ie) {
+      LOG.warn("Could not obtain job info files : " + ie.getMessage(), ie);
+    }
+  }
+
+  private Path getInfoFilePath(String jobId) {
+    return new Path(jobInfoDir, jobId + ".info");
+  }
+  
+  /**
+   * Persists a job in DFS.
+   *
+   * @param job the job about to be 'retired'
+   */
+  public void store(JobInProgress job) {
+    if (active && retainTime > 0) {
+      String jobId = job.getStatus().getJobId();
+      Path jobStatusFile = getInfoFilePath(jobId);
+      try {
+        FSDataOutputStream dataOut = fs.create(jobStatusFile);
+
+        job.getStatus().write(dataOut);
+
+        job.getProfile().write(dataOut);
+
+        job.getCounters().write(dataOut);
+
+        TaskCompletionEvent[] events = 
+                job.getTaskCompletionEvents(0, Integer.MAX_VALUE);
+        dataOut.writeInt(events.length);
+        for (TaskCompletionEvent event : events) {
+          event.write(dataOut);
+        }
+
+        dataOut.close();
+      } catch (IOException ex) {
+        LOG.warn("Could not store [" + jobId + "] job info : " +
+                 ex.getMessage(), ex);
+        try {
+          fs.delete(jobStatusFile);
+        }
+        catch (IOException ex1) {
+          //ignore
+        }
+      }
+    }
+  }
+
+  private FSDataInputStream getJobInfoFile(String jobId) throws IOException {
+    Path jobStatusFile = getInfoFilePath(jobId);
+    return (fs.exists(jobStatusFile)) ? fs.open(jobStatusFile) : null;
+  }
+
+  private JobStatus readJobStatus(FSDataInputStream dataIn) throws IOException {
+    JobStatus jobStatus = new JobStatus();
+    jobStatus.readFields(dataIn);
+    return jobStatus;
+  }
+
+  private JobProfile readJobProfile(FSDataInputStream dataIn)
+          throws IOException {
+    JobProfile jobProfile = new JobProfile();
+    jobProfile.readFields(dataIn);
+    return jobProfile;
+  }
+
+  private Counters readCounters(FSDataInputStream dataIn) throws IOException {
+    Counters counters = new Counters();
+    counters.readFields(dataIn);
+    return counters;
+  }
+
+  private TaskCompletionEvent[] readEvents(FSDataInputStream dataIn,
+                                           int offset, int len)
+          throws IOException {
+    int size = dataIn.readInt();
+    if (offset > size) {
+      return TaskCompletionEvent.EMPTY_ARRAY;
+    }
+    if (offset + len > size) {
+      len = size - offset;
+    }
+    TaskCompletionEvent[] events = new TaskCompletionEvent[len];
+    for (int i = 0; i < (offset + len); i++) {
+      TaskCompletionEvent event = new TaskCompletionEvent();
+      event.readFields(dataIn);
+      if (i >= offset) {
+        events[i - offset] = event;
+      }
+    }
+    return events;
+  }
+
+  /**
+   * This method retrieves JobStatus information from DFS stored using
+   * store method.
+   *
+   * @param jobId the jobId for which jobStatus is queried
+   * @return JobStatus object, null if not able to retrieve
+   */
+  public JobStatus readJobStatus(String jobId) {
+    JobStatus jobStatus = null;
+    if (active) {
+      try {
+        FSDataInputStream dataIn = getJobInfoFile(jobId);
+        if (dataIn != null) {
+          jobStatus = readJobStatus(dataIn);
+          dataIn.close();
+        }
+      } catch (IOException ex) {
+        LOG.warn("Could not read [" + jobId + "] job status : " + ex, ex);
+      }
+    }
+    return jobStatus;
+  }
+
+  /**
+   * This method retrieves JobProfile information from DFS stored using
+   * store method.
+   *
+   * @param jobId the jobId for which jobProfile is queried
+   * @return JobProfile object, null if not able to retrieve
+   */
+  public JobProfile readJobProfile(String jobId) {
+    JobProfile jobProfile = null;
+    if (active) {
+      try {
+        FSDataInputStream dataIn = getJobInfoFile(jobId);
+        if (dataIn != null) {
+          readJobStatus(dataIn);
+          jobProfile = readJobProfile(dataIn);
+          dataIn.close();
+        }
+      } catch (IOException ex) {
+        LOG.warn("Could not read [" + jobId + "] job profile : " + ex, ex);
+      }
+    }
+    return jobProfile;
+  }
+
+  /**
+   * This method retrieves Counters information from DFS stored using
+   * store method.
+   *
+   * @param jobId the jobId for which Counters is queried
+   * @return Counters object, null if not able to retrieve
+   */
+  public Counters readCounters(String jobId) {
+    Counters counters = null;
+    if (active) {
+      try {
+        FSDataInputStream dataIn = getJobInfoFile(jobId);
+        if (dataIn != null) {
+          readJobStatus(dataIn);
+          readJobProfile(dataIn);
+          counters = readCounters(dataIn);
+          dataIn.close();
+        }
+      } catch (IOException ex) {
+        LOG.warn("Could not read [" + jobId + "] job counters : " + ex, ex);
+      }
+    }
+    return counters;
+  }
+
+  /**
+   * This method retrieves TaskCompletionEvents information from DFS stored
+   * using store method.
+   *
+   * @param jobId       the jobId for which TaskCompletionEvents is queried
+   * @param fromEventId events offset
+   * @param maxEvents   max number of events
+   * @return TaskCompletionEvent[], empty array if not able to retrieve
+   */
+  public TaskCompletionEvent[] readJobTaskCompletionEvents(String jobId,
+                                                               int fromEventId,
+                                                               int maxEvents) {
+    TaskCompletionEvent[] events = TaskCompletionEvent.EMPTY_ARRAY;
+    if (active) {
+      try {
+        FSDataInputStream dataIn = getJobInfoFile(jobId);
+        if (dataIn != null) {
+          readJobStatus(dataIn);
+          readJobProfile(dataIn);
+          readCounters(dataIn);
+          events = readEvents(dataIn, fromEventId, maxEvents);
+          dataIn.close();
+        }
+      } catch (IOException ex) {
+        LOG.warn("Could not read [" + jobId + "] job events : " + ex, ex);
+      }
+    }
+    return events;
+  }
+
+}

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

@@ -563,7 +563,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
   ExpireLaunchingTasks expireLaunchingTasks = new ExpireLaunchingTasks();
   Thread expireLaunchingTaskThread = new Thread(expireLaunchingTasks,
                                                 "expireLaunchingTasks");
-    
+
+  CompletedJobStatusStore completedJobStatusStore = null;
+  Thread completedJobsStoreThread = null;
+
   /**
    * It might seem like a bug to maintain a TreeSet of status objects,
    * which can be updated at any time.  But that's not what happens!  We
@@ -701,6 +704,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     synchronized (this) {
       state = State.RUNNING;
     }
+
+    //initializes the job status store
+    completedJobStatusStore = new CompletedJobStatusStore(conf);
+
     LOG.info("Starting RUNNING");
   }
 
@@ -726,6 +733,12 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     this.taskCommitThread = new TaskCommitQueue();
     this.taskCommitThread.start();
 
+    if (completedJobStatusStore.isActive()) {
+      completedJobsStoreThread = new Thread(completedJobStatusStore,
+                                            "completedjobsStore-housekeeper");
+      completedJobsStoreThread.start();
+    }
+
     this.interTrackerServer.join();
     LOG.info("Stopped interTrackerServer");
   }
@@ -788,6 +801,16 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
         ex.printStackTrace();
       }
     }
+    if (this.completedJobsStoreThread != null &&
+        this.completedJobsStoreThread.isAlive()) {
+      LOG.info("Stopping completedJobsStore thread");
+      this.completedJobsStoreThread.interrupt();
+      try {
+        this.completedJobsStoreThread.join();
+      } catch (InterruptedException ex) {
+        ex.printStackTrace();
+      }
+    }
     LOG.info("stopped all jobtracker services");
     return;
   }
@@ -895,7 +918,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
         LOG.info("Removed completed task '" + taskid + "' from '" + 
                  taskTracker + "'");
       }
-      // Clear 
+      // Clear
       trackerToMarkedTasksMap.remove(taskTracker);
     }
   }
@@ -937,6 +960,9 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     // Mark the 'non-running' tasks for pruning
     markCompletedJob(job);
 
+    //persists the job info in DFS
+    completedJobStatusStore.store(job);
+    
     JobEndNotifier.registerNotification(job.getJobConf(), job.getStatus());
 
     // Purge oldest jobs and keep at-most MAX_COMPLETE_USER_JOBS_IN_MEMORY jobs of a given user
@@ -1611,7 +1637,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     if (job != null) {
       return job.getProfile();
     } else {
-      return null;
+      return completedJobStatusStore.readJobProfile(jobid);
     }
   }
   public synchronized JobStatus getJobStatus(String jobid) {
@@ -1619,7 +1645,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     if (job != null) {
       return job.getStatus();
     } else {
-      return null;
+      return completedJobStatusStore.readJobStatus(jobid);
     }
   }
   public synchronized Counters getJobCounters(String jobid) {
@@ -1627,7 +1653,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
     if (job != null) {
       return job.getCounters();
     } else {
-      return null;
+      return completedJobStatusStore.readCounters(jobid);
     }
   }
   public synchronized TaskReport[] getMapTaskReports(String jobid) {
@@ -1679,11 +1705,14 @@ public class JobTracker implements MRConstants, InterTrackerProtocol, JobSubmiss
    */
   public synchronized TaskCompletionEvent[] getTaskCompletionEvents(
                                                                     String jobid, int fromEventId, int maxEvents) throws IOException{
-    TaskCompletionEvent[] events = TaskCompletionEvent.EMPTY_ARRAY;
+    TaskCompletionEvent[] events;
     JobInProgress job = this.jobs.get(jobid);
     if (null != job) {
       events = job.getTaskCompletionEvents(fromEventId, maxEvents);
     }
+    else {
+      events = completedJobStatusStore.readJobTaskCompletionEvents(jobid, fromEventId, maxEvents);
+    }
     return events;
   }
 

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

@@ -0,0 +1,178 @@
+package org.apache.hadoop.mapred;
+
+import junit.framework.TestCase;
+import org.apache.hadoop.dfs.MiniDFSCluster;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Test case to run a MapReduce job.
+ * <p/>
+ * It runs a 2 node cluster Hadoop with a 2 node DFS.
+ * <p/>
+ * The JobConf to use must be obtained via the creatJobConf() method.
+ * <p/>
+ * It creates a temporary directory -accessible via getTestRootDir()-
+ * for both input and output.
+ * <p/>
+ * The input directory is accesible via getInputDir() and the output
+ * directory via getOutputDir()
+ * <p/>
+ * The DFS filesystem is formated before the testcase starts and after it ends.
+ */
+public abstract class ClusterMapReduceTestCase extends TestCase {
+  private MiniDFSCluster dfsCluster = null;
+  private MiniMRCluster mrCluster = null;
+  private FileSystem fileSystem = null;
+
+  /**
+   * Creates Hadoop Cluster and DFS before a test case is run.
+   *
+   * @throws Exception
+   */
+  protected void setUp() throws Exception {
+    super.setUp();
+
+    startCluster(true, null);
+  }
+
+  /**
+   * Starts the cluster within a testcase.
+   * <p/>
+   * Note that the cluster is already started when the testcase method
+   * is invoked. This method is useful if as part of the testcase the
+   * cluster has to be shutdown and restarted again.
+   * <p/>
+   * If the cluster is already running this method does nothing.
+   *
+   * @param reformatDFS indicates if DFS has to be reformated
+   * @param props configuration properties to inject to the mini cluster
+   * @throws Exception if the cluster could not be started
+   */
+  protected synchronized void startCluster(boolean reformatDFS, Properties props)
+          throws Exception {
+    if (dfsCluster == null) {
+      JobConf conf = new JobConf();
+      if (props != null) {
+        for (Map.Entry entry : props.entrySet()) {
+          conf.set((String) entry.getKey(), (String) entry.getValue());
+        }
+      }
+      dfsCluster = new MiniDFSCluster(conf, 2, reformatDFS, null);
+      fileSystem = dfsCluster.getFileSystem();
+
+      ConfigurableMiniMRCluster.setConfiguration(props);
+      //noinspection deprecation
+      mrCluster = new ConfigurableMiniMRCluster(2, fileSystem.getName(), 1);
+    }
+  }
+
+  private static class ConfigurableMiniMRCluster extends MiniMRCluster {
+    private static Properties config;
+
+    public static void setConfiguration(Properties props) {
+      config = props;
+    }
+
+    public ConfigurableMiniMRCluster(int numTaskTrackers, String namenode,
+                                     int numDir) throws Exception {
+      super(numTaskTrackers, namenode, numDir);
+    }
+
+    public JobConf createJobConf() {
+      JobConf conf = super.createJobConf();
+      if (config != null) {
+        for (Map.Entry entry : config.entrySet()) {
+          conf.set((String) entry.getKey(), (String) entry.getValue());
+        }
+      }
+      return conf;
+    }
+  }
+
+  /**
+   * Stops the cluster within a testcase.
+   * <p/>
+   * Note that the cluster is already started when the testcase method
+   * is invoked. This method is useful if as part of the testcase the
+   * cluster has to be shutdown.
+   * <p/>
+   * If the cluster is already stopped this method does nothing.
+   *
+   * @throws Exception if the cluster could not be stopped
+   */
+  protected void stopCluster() throws Exception {
+    if (mrCluster != null) {
+      mrCluster.shutdown();
+      mrCluster = null;
+    }
+    if (dfsCluster != null) {
+      dfsCluster.shutdown();
+      dfsCluster = null;
+      fileSystem = null;
+    }
+  }
+
+  /**
+   * Destroys Hadoop Cluster and DFS after a test case is run.
+   *
+   * @throws Exception
+   */
+  protected void tearDown() throws Exception {
+    stopCluster();
+    super.tearDown();
+  }
+
+  /**
+   * Returns a preconfigured Filesystem instance for test cases to read and
+   * write files to it.
+   * <p/>
+   * TestCases should use this Filesystem instance.
+   *
+   * @return the filesystem used by Hadoop.
+   */
+  protected FileSystem getFileSystem() {
+    return fileSystem;
+  }
+
+  /**
+   * Returns the path to the root directory for the testcase.
+   *
+   * @return path to the root directory for the testcase.
+   */
+  protected Path getTestRootDir() {
+    return new Path("x").getParent();
+  }
+
+  /**
+   * Returns a path to the input directory for the testcase.
+   *
+   * @return path to the input directory for the tescase.
+   */
+  protected Path getInputDir() {
+    return new Path("input");
+  }
+
+  /**
+   * Returns a path to the output directory for the testcase.
+   *
+   * @return path to the output directory for the tescase.
+   */
+  protected Path getOutputDir() {
+    return new Path("output");
+  }
+
+  /**
+   * Returns a job configuration preconfigured to run against the Hadoop
+   * managed by the testcase.
+   *
+   * @return configuration that works on the testcase Hadoop instance
+   */
+  protected JobConf createJobConf() {
+    return mrCluster.createJobConf();
+  }
+
+}

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

@@ -0,0 +1,139 @@
+package org.apache.hadoop.mapred;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+
+import java.io.*;
+import java.util.Iterator;
+import java.util.Properties;
+
+public class TestClusterMapReduceTestCase extends ClusterMapReduceTestCase {
+
+  public static class EchoMap implements Mapper {
+
+    public void configure(JobConf conf) {
+    }
+
+    public void close() {
+    }
+
+    public void map(WritableComparable key, Writable value,
+                    OutputCollector collector, Reporter reporter) throws IOException {
+      collector.collect(key, value);
+    }
+  }
+
+  public static class EchoReduce implements Reducer {
+
+    public void configure(JobConf conf) {
+    }
+
+    public void close() {
+    }
+
+    public void reduce(WritableComparable key, Iterator values,
+                       OutputCollector collector, Reporter reporter) throws IOException {
+      while (values.hasNext()) {
+        Writable value = (Writable) values.next();
+        collector.collect(key, value);
+      }
+    }
+
+  }
+
+  public void _testMapReduce(boolean restart) throws Exception {
+    OutputStream os = getFileSystem().create(new Path(getInputDir(), "text.txt"));
+    Writer wr = new OutputStreamWriter(os);
+    wr.write("hello1\n");
+    wr.write("hello2\n");
+    wr.write("hello3\n");
+    wr.write("hello4\n");
+    wr.close();
+
+    if (restart) {
+      stopCluster();
+      startCluster(false, null);
+    }
+    
+    JobConf conf = createJobConf();
+    conf.setJobName("mr");
+
+    conf.setInputFormat(TextInputFormat.class);
+
+    conf.setMapOutputKeyClass(LongWritable.class);
+    conf.setMapOutputValueClass(Text.class);
+
+    conf.setOutputFormat(TextOutputFormat.class);
+    conf.setOutputKeyClass(LongWritable.class);
+    conf.setOutputValueClass(Text.class);
+
+    conf.setMapperClass(TestClusterMapReduceTestCase.EchoMap.class);
+    conf.setReducerClass(TestClusterMapReduceTestCase.EchoReduce.class);
+
+    conf.setInputPath(getInputDir());
+
+    conf.setOutputPath(getOutputDir());
+
+
+    JobClient.runJob(conf);
+
+    Path[] outputFiles = getFileSystem().listPaths(getOutputDir());
+
+    if (outputFiles.length > 0) {
+      InputStream is = getFileSystem().open(outputFiles[0]);
+      BufferedReader reader = new BufferedReader(new InputStreamReader(is));
+      String line = reader.readLine();
+      int counter = 0;
+      while (line != null) {
+        counter++;
+        assertTrue(line.contains("hello"));
+        line = reader.readLine();
+      }
+      reader.close();
+      assertEquals(4, counter);
+    }
+
+  }
+
+  public void testMapReduce() throws Exception {
+    _testMapReduce(false);
+  }
+
+  public void testMapReduceRestarting() throws Exception {
+    _testMapReduce(true);
+  }
+
+  public void testDFSRestart() throws Exception {
+    Path file = new Path(getInputDir(), "text.txt");
+    OutputStream os = getFileSystem().create(file);
+    Writer wr = new OutputStreamWriter(os);
+    wr.close();
+
+    stopCluster();
+    startCluster(false, null);
+    assertTrue(getFileSystem().exists(file));
+
+    stopCluster();
+    startCluster(true, null);
+    assertFalse(getFileSystem().exists(file));
+    
+  }
+
+  public void testMRConfig() throws Exception {
+    JobConf conf = createJobConf();
+    assertNull(conf.get("xyz"));
+
+    Properties config = new Properties();
+    config.setProperty("xyz", "XYZ");
+    stopCluster();
+    startCluster(false, config);
+
+    conf = createJobConf();
+    assertEquals("XYZ", conf.get("xyz"));
+  }
+
+}

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

@@ -0,0 +1,124 @@
+package org.apache.hadoop.mapred;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.fs.Path;
+
+import java.io.*;
+import java.util.Iterator;
+import java.util.Properties;
+
+public class TestJobStatusPersistency extends ClusterMapReduceTestCase {
+
+  public static class EchoMap implements Mapper {
+
+    public void configure(JobConf conf) {
+    }
+
+    public void close() {
+    }
+
+    public void map(WritableComparable key, Writable value,
+                    OutputCollector collector, Reporter reporter) throws IOException {
+      collector.collect(key, value);
+    }
+  }
+
+  public static class EchoReduce implements Reducer {
+
+    public void configure(JobConf conf) {
+    }
+
+    public void close() {
+    }
+
+    public void reduce(WritableComparable key, Iterator values,
+                       OutputCollector collector, Reporter reporter) throws IOException {
+      while (values.hasNext()) {
+        Writable value = (Writable) values.next();
+        collector.collect(key, value);
+      }
+    }
+
+  }
+
+  private String runJob() throws Exception {
+    OutputStream os = getFileSystem().create(new Path(getInputDir(), "text.txt"));
+    Writer wr = new OutputStreamWriter(os);
+    wr.write("hello1\n");
+    wr.write("hello2\n");
+    wr.write("hello3\n");
+    wr.write("hello4\n");
+    wr.close();
+
+    JobConf conf = createJobConf();
+    conf.setJobName("mr");
+
+    conf.setInputFormat(TextInputFormat.class);
+
+    conf.setMapOutputKeyClass(LongWritable.class);
+    conf.setMapOutputValueClass(Text.class);
+
+    conf.setOutputFormat(TextOutputFormat.class);
+    conf.setOutputKeyClass(LongWritable.class);
+    conf.setOutputValueClass(Text.class);
+
+    conf.setMapperClass(TestJobStatusPersistency.EchoMap.class);
+    conf.setReducerClass(TestJobStatusPersistency.EchoReduce.class);
+
+    conf.setInputPath(getInputDir());
+
+    conf.setOutputPath(getOutputDir());
+
+    return JobClient.runJob(conf).getJobID();
+  }
+
+  public void testNonPersistency() throws Exception {
+    String jobId = runJob();
+    JobClient jc = new JobClient(createJobConf());
+    RunningJob rj = jc.getJob(jobId);
+    assertNotNull(rj);
+    stopCluster();
+    startCluster(false, null);
+    jc = new JobClient(createJobConf());
+    rj = jc.getJob(jobId);
+    assertNull(rj);
+  }
+
+  public void testPersistency() throws Exception {
+    Properties config = new Properties();
+    config.setProperty("mapred.job.tracker.persist.jobstatus.active", "true");
+    config.setProperty("mapred.job.tracker.persist.jobstatus.hours", "1");
+    stopCluster();
+    startCluster(false, config);
+    String jobId = runJob();
+    JobClient jc = new JobClient(createJobConf());
+    RunningJob rj0 = jc.getJob(jobId);
+    assertNotNull(rj0);
+    boolean sucessfull0 = rj0.isSuccessful();
+    String jobName0 = rj0.getJobName();
+    Counters counters0 = rj0.getCounters();
+    TaskCompletionEvent[] events0 = rj0.getTaskCompletionEvents(0);
+
+    stopCluster();
+    startCluster(false, config);
+     
+    jc = new JobClient(createJobConf());
+    RunningJob rj1 = jc.getJob(jobId);
+    assertNotNull(rj1);
+    assertEquals(sucessfull0, rj1.isSuccessful());
+    assertEquals(jobName0, rj0.getJobName());
+    assertEquals(counters0.size(), rj1.getCounters().size());
+
+    TaskCompletionEvent[] events1 = rj1.getTaskCompletionEvents(0);
+    assertEquals(events0.length, events1.length);    
+    for (int i = 0; i < events0.length; i++) {
+      assertEquals(events0[i].getTaskId(), events1[i].getTaskId());
+      assertEquals(events0[i].getTaskStatus(), events1[i].getTaskStatus());
+    }
+  }
+
+}