HADOOP-5394. JobTracker might schedule 2 attempts of the same task with the same attempt id across restarts. Contributed by Amar kamat.

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

CHANGES.txt

@@ -863,6 +863,9 @@ Release 0.20.0 - Unreleased
     HADOOP-5585. Clear FileSystem statistics between tasks when jvm-reuse
     is enabled. (omalley)
 
+    HADOOP-5394. JobTracker might schedule 2 attempts of the same task 
+    with the same attempt id across restarts. (Amar Kamat via sharad)
+
 Release 0.19.2 - Unreleased
 
   BUG FIXES

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

@@ -122,7 +122,7 @@ public class JobHistory {
     FINISHED_MAPS, FINISHED_REDUCES, JOB_STATUS, TASKID, HOSTNAME, TASK_TYPE, 
     ERROR, TASK_ATTEMPT_ID, TASK_STATUS, COPY_PHASE, SORT_PHASE, REDUCE_PHASE, 
     SHUFFLE_FINISHED, SORT_FINISHED, COUNTERS, SPLITS, JOB_PRIORITY, HTTP_PORT, 
-    TRACKER_NAME, STATE_STRING, VERSION, RESTART_COUNT
+    TRACKER_NAME, STATE_STRING, VERSION
   }
 
   /**
@@ -1169,9 +1169,15 @@ public class JobHistory {
      * @param submitTime job's submit time
      * @param launchTime job's launch time
      * @param restartCount number of times the job got restarted
+     * @deprecated Use {@link #logJobInfo(JobID, long, long)} instead.
      */
     public static void logJobInfo(JobID jobid, long submitTime, long launchTime,
                                   int restartCount){
+      logJobInfo(jobid, submitTime, launchTime);
+    }
+
+    public static void logJobInfo(JobID jobid, long submitTime, long launchTime)
+    {
       if (!disableHistory){
         String logFileKey =  JOBTRACKER_UNIQUE_STRING + jobid; 
         ArrayList<PrintWriter> writer = openJobs.get(logFileKey); 
@@ -1179,11 +1185,10 @@ public class JobHistory {
         if (null != writer){
           JobHistory.log(writer, RecordTypes.Job,
                          new Keys[] {Keys.JOBID, Keys.SUBMIT_TIME, 
-                                     Keys.LAUNCH_TIME, Keys.RESTART_COUNT},
+                                     Keys.LAUNCH_TIME},
                          new String[] {jobid.toString(), 
                                        String.valueOf(submitTime), 
-                                       String.valueOf(launchTime),
-                                       String.valueOf(restartCount)});
+                                       String.valueOf(launchTime)});
         }
       }
     }

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

@@ -164,7 +164,7 @@ class JobInProgress {
   long finishTime;
   
   // Indicates how many times the job got restarted
-  private int restartCount = 0;
+  private final int restartCount;
 
   private JobConf conf;
   AtomicBoolean tasksInited = new AtomicBoolean(false);
@@ -214,6 +214,7 @@ class JobInProgress {
     this.maxLevel = NetworkTopology.DEFAULT_HOST_LEVEL;
     this.anyCacheLevel = this.maxLevel+1;
     this.jobtracker = null;
+    this.restartCount = 0;
   }
   
   /**
@@ -222,6 +223,12 @@ class JobInProgress {
    */
   public JobInProgress(JobID jobid, JobTracker jobtracker, 
                        JobConf default_conf) throws IOException {
+    this(jobid, jobtracker, default_conf, 0);
+  }
+  
+  public JobInProgress(JobID jobid, JobTracker jobtracker, 
+                       JobConf default_conf, int rCount) throws IOException {
+    this.restartCount = rCount;
     this.jobId = jobid;
     String url = "http://" + jobtracker.getJobTrackerMachine() + ":" 
         + jobtracker.getInfoPort() + "/jobdetails.jsp?jobid=" + jobid;
@@ -595,19 +602,17 @@ class JobInProgress {
   }
 
   // Update the job start/launch time (upon restart) and log to history
-  synchronized void updateJobInfo(long startTime, long launchTime, int count) {
+  synchronized void updateJobInfo(long startTime, long launchTime) {
     // log and change to the job's start/launch time
     this.startTime = startTime;
     this.launchTime = launchTime;
-    // change to the job's restart count
-    this.restartCount = count;
-    JobHistory.JobInfo.logJobInfo(jobId, startTime, launchTime, count);
+    JobHistory.JobInfo.logJobInfo(jobId, startTime, launchTime);
   }
 
   /**
    * Get the number of times the job has restarted
    */
-  int numRestarts() {
+  int getNumRestarts() {
     return restartCount;
   }
   

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

@@ -46,6 +46,8 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -123,6 +125,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
   final static FsPermission SYSTEM_DIR_PERMISSION =
     FsPermission.createImmutable((short) 0733); // rwx-wx-wx
 
+  // system files should have 700 permission
+  final static FsPermission SYSTEM_FILE_PERMISSION =
+    FsPermission.createImmutable((short) 0700); // rwx------
+
   /**
    * A client tried to submit a job before the Job Tracker was ready.
    */
@@ -672,6 +678,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     Set<JobID> jobsToRecover; // set of jobs to be recovered
     
     private int totalEventsRecovered = 0;
+    private int restartCount = 0;
+    private boolean shouldRecover = false;
 
     Set<String> recoveredTrackers = 
       Collections.synchronizedSet(new HashSet<String>());
@@ -850,7 +858,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     }
 
     public boolean shouldRecover() {
-      return jobsToRecover.size() != 0;
+      return shouldRecover;
     }
 
     public boolean shouldSchedule() {
@@ -888,18 +896,16 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     
     // checks if the job dir has the required files
     public void checkAndAddJob(FileStatus status) throws IOException {
-      String jobName = status.getPath().getName();
-      if (isJobNameValid(jobName)) {
+      String fileName = status.getPath().getName();
+      if (isJobNameValid(fileName)) {
         if (JobClient.isJobDirValid(status.getPath(), fs)) {
-          recoveryManager.addJobForRecovery(JobID.forName(jobName));
+          recoveryManager.addJobForRecovery(JobID.forName(fileName));
+          shouldRecover = true; // enable actual recovery if num-files > 1
         } else {
-          LOG.info("Found an incomplete job directory " + jobName + "." 
+          LOG.info("Found an incomplete job directory " + fileName + "." 
                    + " Deleting it!!");
           fs.delete(status.getPath(), true);
         }
-      } else {
-        LOG.info("Deleting " + status.getPath());
-        fs.delete(status.getPath(), true);
       }
     }
     
@@ -918,8 +924,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
       // Set the start/launch time only if there are recovered tasks
       // Increment the job's restart count
       jip.updateJobInfo(job.getLong(JobHistory.Keys.SUBMIT_TIME), 
-                        job.getLong(JobHistory.Keys.LAUNCH_TIME),
-                        job.getInt(Keys.RESTART_COUNT) + 1);
+                        job.getLong(JobHistory.Keys.LAUNCH_TIME));
 
       // Save the new job status
       JobStatus newStatus = (JobStatus)jip.getStatus().clone();
@@ -1119,7 +1124,84 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
      expireLaunchingTasks.removeTask(attemptId);
     }
   
+    Path getRestartCountFile() {
+      return new Path(getSystemDir(), "jobtracker.info");
+    }
+
+    Path getTempRestartCountFile() {
+      return new Path(getSystemDir(), "jobtracker.info.recover");
+    }
+
+    /**
+     * Initialize the recovery process. It simply creates a jobtracker.info file
+     * in the jobtracker's system directory and writes its restart count in it.
+     * For the first start, the jobtracker writes '0' in it. Upon subsequent 
+     * restarts the jobtracker replaces the count with its current count which 
+     * is (old count + 1). The whole purpose of this api is to obtain restart 
+     * counts across restarts to avoid attempt-id clashes.
+     * 
+     * Note that in between if the jobtracker.info files goes missing then the
+     * jobtracker will disable recovery and continue. 
+     *  
+     */
+    void updateRestartCount() throws IOException {
+      Path restartFile = getRestartCountFile();
+      Path tmpRestartFile = getTempRestartCountFile();
+      FileSystem fs = restartFile.getFileSystem(conf);
+      FsPermission filePerm = new FsPermission(SYSTEM_FILE_PERMISSION);
+
+      // read the count from the jobtracker info file
+      if (fs.exists(restartFile)) {
+        fs.delete(tmpRestartFile, false); // delete the tmp file
+      } else if (fs.exists(tmpRestartFile)) {
+        // if .rec exists then delete the main file and rename the .rec to main
+        fs.rename(tmpRestartFile, restartFile); // rename .rec to main file
+      } else {
+        // For the very first time the jobtracker will create a jobtracker.info
+        // file. If the jobtracker has restarted then disable recovery as files'
+        // needed for recovery are missing.
+
+        // disable recovery if this is a restart
+        shouldRecover = false;
+
+        // write the jobtracker.info file
+        FSDataOutputStream out = FileSystem.create(fs, restartFile, filePerm);
+        out.writeInt(0);
+        out.close();
+        return;
+      }
+
+      FSDataInputStream in = fs.open(restartFile);
+      // read the old count
+      restartCount = in.readInt();
+      ++restartCount; // increment the restart count
+      in.close();
+
+      // Write back the new restart count and rename the old info file
+      //TODO This is similar to jobhistory recovery, maybe this common code
+      //      can be factored out.
+      
+      // write to the tmp file
+      FSDataOutputStream out = FileSystem.create(fs, tmpRestartFile, filePerm);
+      out.writeInt(restartCount);
+      out.close();
+
+      // delete the main file
+      fs.delete(restartFile, false);
+      
+      // rename the .rec to main file
+      fs.rename(tmpRestartFile, restartFile);
+    }
+
     public void recover() {
+      if (!shouldRecover()) {
+        // clean up jobs structure
+        jobsToRecover.clear();
+        return;
+      }
+
+      LOG.info("Restart count of the jobtracker : " + restartCount);
+
       // I. Init the jobs and cache the recovered job history filenames
       Map<JobID, Path> jobHistoryFilenameMap = new HashMap<JobID, Path>();
       Iterator<JobID> idIter = jobsToRecover.iterator();
@@ -1128,7 +1210,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         LOG.info("Trying to recover details of job " + id);
         try {
           // 1. Create the job object
-          JobInProgress job = new JobInProgress(id, JobTracker.this, conf);
+          JobInProgress job = 
+            new JobInProgress(id, JobTracker.this, conf, restartCount);
 
           // 2. Check if the user has appropriate access
           // Get the user group info for the job's owner
@@ -1209,8 +1292,6 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         // 3. Close the listener
         listener.close();
         
-        LOG.info("Restart count for job " + id + " is " + pJob.numRestarts());
-
         // 4. Update the recovery metric
         totalEventsRecovered += listener.getNumEventsRecovered();
 
@@ -1532,7 +1613,6 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         }
         // Make sure that the backup data is preserved
         FileStatus[] systemDirData = fs.listStatus(this.systemDir);
-        LOG.info("Cleaning up the system directory");
         // Check if the history is enabled .. as we cant have persistence with 
         // history disabled
         if (conf.getBoolean("mapred.jobtracker.restart.recover", false) 
@@ -1553,6 +1633,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
             break; // if there is something to recover else clean the sys dir
           }
         }
+        LOG.info("Cleaning up the system directory");
         fs.delete(systemDir, true);
         if (FileSystem.mkdirs(fs, systemDir, 
             new FsPermission(SYSTEM_DIR_PERMISSION))) {
@@ -1569,6 +1650,24 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
       }
       Thread.sleep(SYSTEM_DIR_CLEANUP_RETRY_PERIOD);
     }
+
+    // Prepare for recovery. This is done irrespective of the status of restart
+    // flag.
+    try {
+      recoveryManager.updateRestartCount();
+    } catch (IOException ioe) {
+      LOG.warn("Failed to initialize recovery manager. The Recovery manager "
+               + "failed to access the system files in the system dir (" 
+               + getSystemDir() + ")."); 
+      LOG.warn("It might be because the JobTracker failed to read/write system"
+               + " files (" + recoveryManager.getRestartCountFile() + " / " 
+               + recoveryManager.getTempRestartCountFile() + ") or the system "
+               + " file " + recoveryManager.getRestartCountFile() 
+               + " is missing!");
+      LOG.warn("Bailing out...");
+      throw ioe;
+    }
+    
     // Same with 'localDir' except it's always on the local disk.
     jobConf.deleteLocalFiles(SUBDIR);
 

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

@@ -887,7 +887,7 @@ class TaskInProgress {
     TaskAttemptID taskid = null;
     if (nextTaskId < (MAX_TASK_EXECS + maxTaskAttempts + numKilledTasks)) {
       // Make sure that the attempts are unqiue across restarts
-      int attemptId = job.numRestarts() * NUM_ATTEMPTS_PER_RESTART + nextTaskId;
+      int attemptId = job.getNumRestarts() * NUM_ATTEMPTS_PER_RESTART + nextTaskId;
       taskid = new TaskAttemptID( id, attemptId);
       ++nextTaskId;
     } else {

src/test/org/apache/hadoop/hdfs/TestHDFSServerPorts.java

@@ -66,6 +66,29 @@ public class TestHDFSServerPorts extends TestCase {
     return NameNode.createNameNode(args, config);
   }
 
+  /**
+   * Start the data-node.
+   */
+  public DataNode startDataNode(int index, Configuration config) 
+  throws IOException {
+    String dataDir = System.getProperty("test.build.data");
+    File dataNodeDir = new File(dataDir, "data-" + index);
+    config.set("dfs.data.dir", dataNodeDir.getPath());
+
+    String[] args = new String[] {};
+    // NameNode will modify config with the ports it bound to
+    return DataNode.createDataNode(args, config);
+  }
+
+  /**
+   * Stop the datanode.
+   */
+  public void stopDataNode(DataNode dn) {
+    if (dn != null) {
+      dn.shutdown();
+    }
+  }
+
   public void stopNameNode(NameNode nn) {
     if (nn != null) {
       nn.stop();

src/test/org/apache/hadoop/ipc/TestSocketFactory.java

@@ -85,6 +85,7 @@ public class TestSocketFactory extends TestCase {
       assertTrue(dfs.exists(filePath));
 
       // This will test TPC to a JobTracker
+      fs = FileSystem.get(sconf);
       mr = new MiniMRCluster(1, fs.getUri().toString(), 1);
       final int jobTrackerPort = mr.getJobTrackerPort();
 

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

@@ -543,24 +543,28 @@ public class MiniMRCluster {
     jobTrackerThread = new Thread(jobTracker);
         
     jobTrackerThread.start();
-    while (!jobTracker.isUp()) {
+    while (jobTracker.isActive() && !jobTracker.isUp()) {
       try {                                     // let daemons get started
         Thread.sleep(1000);
       } catch(InterruptedException e) {
       }
     }
         
-    ClusterStatus status = jobTracker.getJobTracker().getClusterStatus(false);
-    while (jobTracker.isActive() && status.getJobTrackerState() == JobTracker.State.INITIALIZING) {
-      try {
-        LOG.info("JobTracker still initializing. Waiting.");
-        Thread.sleep(1000);
-      } catch(InterruptedException e) {}
+    // is the jobtracker has started then wait for it to init
+    ClusterStatus status = null;
+    if (jobTracker.isUp()) {
       status = jobTracker.getJobTracker().getClusterStatus(false);
+      while (jobTracker.isActive() && status.getJobTrackerState() 
+             == JobTracker.State.INITIALIZING) {
+        try {
+          LOG.info("JobTracker still initializing. Waiting.");
+          Thread.sleep(1000);
+        } catch(InterruptedException e) {}
+        status = jobTracker.getJobTracker().getClusterStatus(false);
+      }
     }
 
-    if (!jobTracker.isActive() 
-        || status.getJobTrackerState() != JobTracker.State.RUNNING) {
+    if (!jobTracker.isActive()) {
       // return if jobtracker has crashed
       return;
     }

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.mapred;
 import java.io.IOException;
 import junit.framework.TestCase;
 import org.apache.hadoop.hdfs.TestHDFSServerPorts;
+import org.apache.hadoop.hdfs.server.datanode.DataNode;
 import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.fs.FileSystem;
 
@@ -112,8 +113,10 @@ public class TestMRServerPorts extends TestCase {
    */
   public void testJobTrackerPorts() throws Exception {
     NameNode nn = null;
+    DataNode dn = null;
     try {
       nn = hdfs.startNameNode();
+      dn = hdfs.startDataNode(1, hdfs.getConfig());
 
       // start job tracker on the same port as name-node
       JobConf conf2 = new JobConf(hdfs.getConfig());
@@ -139,6 +142,7 @@ public class TestMRServerPorts extends TestCase {
       assertTrue(started); // should start now
 
     } finally {
+      hdfs.stopDataNode(dn);
       hdfs.stopNameNode(nn);
     }
   }
@@ -148,10 +152,12 @@ public class TestMRServerPorts extends TestCase {
    */
   public void testTaskTrackerPorts() throws Exception {
     NameNode nn = null;
+    DataNode dn = null;
     JobTracker jt = null;
     JTRunner runner = null;
     try {
       nn = hdfs.startNameNode();
+      dn = hdfs.startDataNode(2, hdfs.getConfig());
 
       JobConf conf2 = new JobConf(hdfs.getConfig());
       runner = new JTRunner();
@@ -187,6 +193,7 @@ public class TestMRServerPorts extends TestCase {
         runner.interrupt();
         runner.join();
       }
+      hdfs.stopDataNode(dn);
       hdfs.stopNameNode(nn);
     }
   }

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

@@ -239,7 +239,7 @@ public class TestQueueManager extends TestCase {
         Path sysDir = new Path(jobtracker.getSystemDir());
         FileSystem fs = sysDir.getFileSystem(conf);
         int size = fs.listStatus(sysDir).length;
-        while (size > 0) {
+        while (size > 1) { // ignore the jobtracker.info file
           System.out.println("Waiting for the job files in sys directory to be cleaned up");
           UtilsForTests.waitFor(100);
           size = fs.listStatus(sysDir).length;

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

@@ -29,6 +29,8 @@ import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.JobTracker.RecoveryManager;
+import org.apache.hadoop.mapred.MiniMRCluster.JobTrackerRunner;
+import org.apache.hadoop.mapred.TestJobInProgressListener.MyScheduler;
 import org.apache.hadoop.security.UserGroupInformation;
 
 /**
@@ -261,4 +263,139 @@ public class TestRecoveryManager extends TestCase {
     
     mr.shutdown();
   }
+  
+  /**
+   * Test if restart count of the jobtracker is correctly managed.
+   * Steps are as follows :
+   *   - start the jobtracker and check if the info file gets created.
+   *   - stops the jobtracker, deletes the jobtracker.info file and checks if
+   *     upon restart the recovery is 'off'
+   *   - submit a job to the jobtracker.
+   *   - restart the jobtracker k times and check if the restart count on ith 
+   *     iteration is i.
+   *   - submit a new job and check if its restart count is 0.
+   *   - garble the jobtracker.info file and restart he jobtracker, the 
+   *     jobtracker should crash.
+   */
+  public void testRestartCount() throws Exception {
+    LOG.info("Testing restart-count");
+    String signalFile = new Path(TEST_DIR, "signal").toString();
+    
+    // clean up
+    FileSystem fs = FileSystem.get(new Configuration());
+    fs.delete(TEST_DIR, true);
+    
+    JobConf conf = new JobConf();
+    conf.set("mapred.jobtracker.job.history.block.size", "1024");
+    conf.set("mapred.jobtracker.job.history.buffer.size", "1024");
+    conf.setBoolean("mapred.jobtracker.restart.recover", true);
+    // since there is no need for initing
+    conf.setClass("mapred.jobtracker.taskScheduler", MyScheduler.class,
+                  TaskScheduler.class);
+    
+    MiniMRCluster mr = new MiniMRCluster(1, "file:///", 1, null, null, conf);
+    JobTracker jobtracker = mr.getJobTrackerRunner().getJobTracker();
+    JobClient jc = new JobClient(mr.createJobConf());
+
+    // check if the jobtracker info file exists
+    Path infoFile = jobtracker.recoveryManager.getRestartCountFile();
+    assertTrue("Jobtracker infomation is missing", fs.exists(infoFile));
+
+    // check if garbling the system files disables the recovery process
+    LOG.info("Stopping jobtracker for testing with system files deleted");
+    mr.stopJobTracker();
+    
+    // delete the info file
+    Path rFile = jobtracker.recoveryManager.getRestartCountFile();
+    fs.delete(rFile,false);
+    
+    // start the jobtracker
+    LOG.info("Stopping jobtracker with system files deleted");
+    mr.startJobTracker();
+    
+    UtilsForTests.waitForJobTracker(jc);
+    jobtracker = mr.getJobTrackerRunner().getJobTracker();
+
+    // check if the recovey is disabled
+    assertFalse("Recovery is not disabled upon missing system files", 
+                jobtracker.recoveryManager.shouldRecover());
+
+    // check if the system dir is sane
+    assertTrue("Recovery file is missing upon restart", fs.exists(rFile));
+    Path tFile = jobtracker.recoveryManager.getTempRestartCountFile();
+    assertFalse("Temp recovery file exists upon restart", fs.exists(tFile));
+
+    // submit a job
+    JobConf job = mr.createJobConf();
+    
+    UtilsForTests.configureWaitingJobConf(job, 
+        new Path(TEST_DIR, "input"), new Path(TEST_DIR, "output6"), 2, 0, 
+        "test-recovery-manager", signalFile, signalFile);
+    
+    // submit the faulty job
+    RunningJob rJob = jc.submitJob(job);
+    LOG.info("Submitted first job " + rJob.getID());
+
+    // kill the jobtracker multiple times and check if the count is correct
+    for (int i = 1; i <= 5; ++i) {
+      LOG.info("Stopping jobtracker for " + i + " time");
+      mr.stopJobTracker();
+      
+      // start the jobtracker
+      LOG.info("Starting jobtracker for " + i + " time");
+      mr.startJobTracker();
+      
+      UtilsForTests.waitForJobTracker(jc);
+      
+      // check if the system dir is sane
+      assertTrue("Recovery file is missing upon restart", fs.exists(rFile));
+      assertFalse("Temp recovery file exists upon restart", fs.exists(tFile));
+      
+      jobtracker = mr.getJobTrackerRunner().getJobTracker();
+      JobInProgress jip = jobtracker.getJob(rJob.getID());
+      
+      // assert if restart count is correct
+      assertEquals("Recovery manager failed to recover restart count",
+                   i, jip.getNumRestarts());
+    }
+    
+    // kill the old job
+    rJob.killJob();
+
+    // II. Submit a new job and check if the restart count is 0
+    JobConf job1 = mr.createJobConf();
+    
+    UtilsForTests.configureWaitingJobConf(job1, 
+        new Path(TEST_DIR, "input"), new Path(TEST_DIR, "output7"), 50, 0, 
+        "test-recovery-manager", signalFile, signalFile);
+    
+    // make sure that the job id's dont clash
+    jobtracker.getNewJobId();
+
+    // submit a new job
+    rJob = jc.submitJob(job1);
+    LOG.info("Submitted first job after restart" + rJob.getID());
+
+    // assert if restart count is correct
+    JobInProgress jip = jobtracker.getJob(rJob.getID());
+    assertEquals("Restart count for new job is incorrect",
+                 0, jip.getNumRestarts());
+
+    LOG.info("Stopping jobtracker for testing the fs errors");
+    mr.stopJobTracker();
+
+    // check if system.dir problems in recovery kills the jobtracker
+    fs.delete(rFile, false);
+    FSDataOutputStream out = fs.create(rFile);
+    out.writeBoolean(true);
+    out.close();
+
+    // start the jobtracker
+    LOG.info("Starting jobtracker with fs errors");
+    mr.startJobTracker();
+    JobTrackerRunner runner = mr.getJobTrackerRunner();
+    assertFalse("Restart count for new job is incorrect", runner.isActive());
+
+    mr.shutdown();
+  } 
 }