hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sha...@apache.org
Subject svn commit: r803231 - in /hadoop/mapreduce/trunk: ./ src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/ src/java/ src/java/org/apache/hadoop/mapred/ src/test/mapred/org/apache/hadoop/mapred/ src/webapps/job/
Date Tue, 11 Aug 2009 17:43:14 GMT
Author: sharad
Date: Tue Aug 11 17:43:14 2009
New Revision: 803231

URL: http://svn.apache.org/viewvc?rev=803231&view=rev
Log:
MAPREDUCE-817. Add a cache for retired jobs with minimal job info and provide a way to access history file url.

Removed:
    hadoop/mapreduce/trunk/src/webapps/job/loadhistory.jsp
Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java
    hadoop/mapreduce/trunk/src/java/mapred-default.xml
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JSPUtil.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobClient.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobHistory.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobStatus.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobSubmissionProtocol.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/RunningJob.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobHistoryParsing.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobQueueTaskScheduler.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java
    hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestParallelInitialization.java
    hadoop/mapreduce/trunk/src/webapps/job/analysejobhistory.jsp
    hadoop/mapreduce/trunk/src/webapps/job/jobconf_history.jsp
    hadoop/mapreduce/trunk/src/webapps/job/jobdetailshistory.jsp
    hadoop/mapreduce/trunk/src/webapps/job/jobtaskshistory.jsp
    hadoop/mapreduce/trunk/src/webapps/job/jobtracker.jsp
    hadoop/mapreduce/trunk/src/webapps/job/taskdetailshistory.jsp

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Tue Aug 11 17:43:14 2009
@@ -17,6 +17,9 @@
     MAPREDUCE-766. Enhanced list-blacklisted-trackers to display reasons
     for blacklisting a node. (Sreekanth Ramakrishnan via yhemanth)
 
+    MAPREDUCE-817. Add a cache for retired jobs with minimal job info and 
+    provide a way to access history file url. (sharad)
+
   NEW FEATURES
 
     MAPREDUCE-546. Provide sample fair scheduler config file in conf/ and use

Modified: hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java (original)
+++ hadoop/mapreduce/trunk/src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java Tue Aug 11 17:43:14 2009
@@ -172,7 +172,9 @@
       super(jId, jobConf, null);
       this.taskTrackerManager = taskTrackerManager;
       this.startTime = System.currentTimeMillis();
-      this.status = new JobStatus(jId, 0f, 0f, JobStatus.PREP);
+      this.status = new JobStatus(jId, 0f, 0f, JobStatus.PREP, 
+          jobConf.getUser(), 
+          jobConf.getJobName(), "", "");
       this.status.setJobPriority(JobPriority.NORMAL);
       this.status.setStartTime(startTime);
       if (null == jobConf.getQueueName()) {

Modified: hadoop/mapreduce/trunk/src/java/mapred-default.xml
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/mapred-default.xml?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/mapred-default.xml (original)
+++ hadoop/mapreduce/trunk/src/java/mapred-default.xml Tue Aug 11 17:43:14 2009
@@ -348,6 +348,21 @@
 </property>
 
 <property>
+  <name>mapred.job.tracker.retiredjobs.cache.size</name>
+  <value>1000</value>
+  <description>The number of retired job status to keep in the cache.
+  </description>
+</property>
+
+<property>
+  <name>mapred.job.tracker.jobhistory.lru.cache.size</name>
+  <value>5</value>
+  <description>The number of job history files loaded in memory. The jobs are 
+  loaded when they are first accessed. The cache is cleared based on LRU.
+  </description>
+</property>
+
+<property>
   <name>mapred.jobtracker.instrumentation</name>
   <value>org.apache.hadoop.mapred.JobTrackerMetricsInst</value>
   <description>Expert: The instrumentation class to associate with each JobTracker.

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JSPUtil.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JSPUtil.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JSPUtil.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JSPUtil.java Tue Aug 11 17:43:14 2009
@@ -18,21 +18,37 @@
 package org.apache.hadoop.mapred;
 
 import java.io.IOException;
-import java.util.Iterator;
+import java.net.URLEncoder;
 import java.util.Collection;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.mapred.JobHistory.JobInfo;
 import org.apache.hadoop.util.ServletUtil;
+import org.apache.hadoop.util.StringUtils;
 
 class JSPUtil {
   private static final String PRIVATE_ACTIONS_KEY = "webinterface.private.actions";
   
   public static final Configuration conf = new Configuration();
 
+  //LRU based cache
+  private static final Map<String, JobInfo> jobHistoryCache = 
+    new LinkedHashMap<String, JobInfo>(); 
+
+  private static final int CACHE_SIZE = 
+    conf.getInt("mapred.job.tracker.jobhistory.lru.cache.size", 5);
+
+  private static final Log LOG = LogFactory.getLog(JSPUtil.class);
   /**
    * Method used to process the request from the job page based on the 
    * request which it has received. For example like changing priority.
@@ -178,7 +194,94 @@
     return sb.toString();
   }
 
+  @SuppressWarnings("unchecked")
+  public static String generateRetiredJobTable(JobTracker tracker, int rowId) 
+    throws IOException {
+
+    StringBuffer sb = new StringBuffer();
+    sb.append("<table border=\"1\" cellpadding=\"5\" cellspacing=\"0\">\n");
+
+    Iterator<JobStatus> iterator = 
+      tracker.retireJobs.getAll().descendingIterator();
+    if (!iterator.hasNext()) {
+      sb.append("<tr><td align=\"center\" colspan=\"8\"><i>none</i>" +
+      "</td></tr>\n");
+    } else {
+      sb.append("<tr>");
+      
+      sb.append("<td><b>Jobid</b></td>");
+      sb.append("<td><b>Priority</b></td>");
+      sb.append("<td><b>User</b></td>");
+      sb.append("<td><b>Name</b></td>");
+      sb.append("<td><b>State</b></td>");
+      sb.append("<td><b>Start Time</b></td>");
+      sb.append("<td><b>Finish Time</b></td>");
+      sb.append("<td><b>Map % Complete</b></td>");
+      sb.append("<td><b>Reduce % Complete</b></td>");
+      sb.append("<td><b>Job Scheduling Information</b></td>");
+      sb.append("</tr>\n");
+      for (int i = 0; i < 100 && iterator.hasNext(); i++) {
+        JobStatus status = iterator.next();
+        sb.append("<tr>");
+        sb.append(
+            "<td id=\"job_" + rowId + "\">" + 
+              "<a href=\"jobdetailshistory.jsp?jobid=" + 
+              status.getJobId() + "&logFile=" + 
+              URLEncoder.encode(status.getHistoryFile().toString(), 
+                "UTF-8") + "\">" + 
+              status.getJobId() + "</a></td>" +
+            
+            "<td id=\"priority_" + rowId + "\">" + 
+              status.getJobPriority().toString() + "</td>" +
+            "<td id=\"user_" + rowId + "\">" + status.getUsername() + "</td>" +
+            "<td id=\"name_" + rowId + "\">" + status.getJobName() + "</td>" +
+            "<td>" + JobStatus.getJobRunState(status.getRunState()) + "</td>" +
+            "<td>" + new Date(status.getStartTime()) + "</td>" +
+            "<td>" + new Date(status.getFinishTime()) + "</td>" +
+            
+            "<td>" + StringUtils.formatPercent(status.mapProgress(), 2)
+            + ServletUtil.percentageGraph(status.mapProgress() * 100, 80) + 
+            "</td>" +
+            
+            "<td>" + StringUtils.formatPercent(status.reduceProgress(), 2)
+            + ServletUtil.percentageGraph(status.reduceProgress() * 100, 80) + 
+            "</td>" +
+            
+            "<td>" + status.getSchedulingInfo() + "</td>" +
+            
+            "</tr>\n");
+        rowId++;
+      }
+    }
+    sb.append("</table>\n");
+    return sb.toString();
+  }
+
   static final boolean privateActionsAllowed() {
     return conf.getBoolean(PRIVATE_ACTIONS_KEY, false);
   }
+
+  static JobInfo getJobInfo(HttpServletRequest request, FileSystem fs) 
+      throws IOException {
+    String jobid = request.getParameter("jobid");
+    String logFile = request.getParameter("logFile");
+    synchronized(jobHistoryCache) {
+      JobInfo jobInfo = jobHistoryCache.remove(jobid);
+      if (jobInfo == null) {
+        jobInfo = new JobHistory.JobInfo(jobid);
+        LOG.info("Loading Job History file "+jobid + ".   Cache size is " +
+            jobHistoryCache.size());
+        DefaultJobHistoryParser.parseJobTasks( logFile, jobInfo, fs) ; 
+      }
+      jobHistoryCache.put(jobid, jobInfo);
+      if (jobHistoryCache.size() > CACHE_SIZE) {
+        Iterator<Map.Entry<String, JobInfo>> it = 
+          jobHistoryCache.entrySet().iterator();
+        String removeJobId = it.next().getKey();
+        it.remove();
+        LOG.info("Job History file removed form cache "+removeJobId);
+      }
+      return jobInfo;
+    }
+  }
 }

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobClient.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobClient.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobClient.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobClient.java Tue Aug 11 17:43:14 2009
@@ -173,7 +173,6 @@
    * remote service to provide certain functionality.
    */
   class NetworkedJob implements RunningJob {
-    JobProfile profile;
     JobStatus status;
     long statustime;
 
@@ -186,7 +185,6 @@
      */
     public NetworkedJob(JobStatus job) throws IOException {
       this.status = job;
-      this.profile = jobSubmitClient.getJobProfile(job.getJobID());
       this.statustime = System.currentTimeMillis();
     }
 
@@ -205,7 +203,10 @@
      * @throws IOException
      */
     synchronized void updateStatus() throws IOException {
-      this.status = jobSubmitClient.getJobStatus(profile.getJobID());
+      this.status = jobSubmitClient.getJobStatus(status.getJobID());
+      if (this.status == null) {
+        throw new IOException("Job status not available ");
+      }
       this.statustime = System.currentTimeMillis();
     }
 
@@ -213,35 +214,35 @@
      * An identifier for the job
      */
     public JobID getID() {
-      return profile.getJobID();
+      return status.getJobID();
     }
     
     /** @deprecated This method is deprecated and will be removed. Applications should 
      * rather use {@link #getID()}.*/
     @Deprecated
     public String getJobID() {
-      return profile.getJobID().toString();
+      return status.getJobID().toString();
     }
     
     /**
      * The user-specified job name
      */
     public String getJobName() {
-      return profile.getJobName();
+      return status.getJobName();
     }
 
     /**
      * The name of the job file
      */
     public String getJobFile() {
-      return profile.getJobFile();
+      return status.getJobFile();
     }
 
     /**
      * A URL where the job's status can be seen
      */
     public String getTrackingURL() {
-      return profile.getURL().toString();
+      return status.getTrackingUrl();
     }
 
     /**
@@ -368,11 +369,14 @@
         updateStatus();
       } catch (IOException e) {
       }
-      return "Job: " + profile.getJobID() + "\n" + 
-        "file: " + profile.getJobFile() + "\n" + 
-        "tracking URL: " + profile.getURL() + "\n" + 
+      return "Job: " + status.getJobID() + "\n" +
+        "status: " + JobStatus.getJobRunState(status.getRunState()) + "\n" + 
+        "file: " + status.getJobFile() + "\n" + 
+        "tracking URL: " + status.getTrackingUrl() + "\n" + 
         "map() completion: " + status.mapProgress() + "\n" + 
-        "reduce() completion: " + status.reduceProgress();
+        "reduce() completion: " + status.reduceProgress() + "\n" +
+        "history URL: " + status.getHistoryFile() + "\n" +
+        "retired: " + status.isRetired();
     }
         
     /**
@@ -386,6 +390,16 @@
     public String[] getTaskDiagnostics(TaskAttemptID id) throws IOException {
       return jobSubmitClient.getTaskDiagnostics(id);
     }
+
+    public String getHistoryUrl() throws IOException {
+      updateStatus();
+      return status.getHistoryFile();
+    }
+
+    public boolean isRetired() throws IOException {
+      updateStatus();
+      return status.isRetired();
+    }
   }
 
   private JobSubmissionProtocol jobSubmitClient;
@@ -1354,7 +1368,10 @@
       }
     }
     LOG.info("Job complete: " + jobId);
-    job.getCounters().log(LOG);
+    Counters counters = job.getCounters();
+    if (counters != null) {
+      counters.log(LOG);
+    }
     return job.isSuccessful();
   }
 
@@ -1695,7 +1712,12 @@
         } else {
           System.out.println();
           System.out.println(job);
-          System.out.println(job.getCounters());
+          Counters counters = job.getCounters();
+          if (counters != null) {
+            System.out.println(counters);
+          } else {
+            System.out.println("Counters not available. Job is retired.");
+          }
           exitCode = 0;
         }
       } else if (getCounter) {
@@ -1704,10 +1726,16 @@
           System.out.println("Could not find job " + jobid);
         } else {
           Counters counters = job.getCounters();
-          Group group = counters.getGroup(counterGroupName);
-          Counter counter = group.getCounterForName(counterName);
-          System.out.println(counter.getCounter());
-          exitCode = 0;
+          if (counters == null) {
+            System.out.println("Counters not available for retired job " + 
+                jobid);
+            exitCode = -1;
+          } else {
+            Group group = counters.getGroup(counterGroupName);
+            Counter counter = group.getCounterForName(counterName);
+            System.out.println(counter.getCounter());
+            exitCode = 0;
+          }
         }
       } else if (killJob) {
         RunningJob job = getJob(JobID.forName(jobid));

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobHistory.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobHistory.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobHistory.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobHistory.java Tue Aug 11 17:43:14 2009
@@ -127,13 +127,16 @@
 
     private ThreadPoolExecutor executor = null;
     private final Configuration conf;
+    private final JobTracker jobTracker;
 
     // cache from job-key to files associated with it.
     private Map<JobID, FilesHolder> fileCache = 
       new ConcurrentHashMap<JobID, FilesHolder>();
 
-    JobHistoryFilesManager(Configuration conf) throws IOException {
+    JobHistoryFilesManager(Configuration conf, JobTracker jobTracker) 
+      throws IOException {
       this.conf = conf;
+      this.jobTracker = jobTracker;
     }
 
     void start() {
@@ -184,7 +187,22 @@
       fileCache.remove(id);
     }
 
-    void moveToDone(final JobID id, final List<Path> paths) {
+    void moveToDone(final JobID id) {
+      final List<Path> paths = new ArrayList<Path>();
+      final Path historyFile = fileManager.getHistoryFile(id);
+      if (historyFile == null) {
+        LOG.info("No file for job-history with " + id + " found in cache!");
+      } else {
+        paths.add(historyFile);
+      }
+
+      final Path confPath = fileManager.getConfFileWriters(id);
+      if (confPath == null) {
+        LOG.info("No file for jobconf with " + id + " found in cache!");
+      } else {
+        paths.add(confPath);
+      }
+
       executor.execute(new Runnable() {
 
         public void run() {
@@ -201,6 +219,13 @@
               }
             }
 
+            String historyFileDonePath = null;
+            if (historyFile != null) {
+              historyFileDonePath = new Path(DONE, 
+                  historyFile.getName()).toString();
+            }
+            jobTracker.historyFileCopied(id, historyFileDonePath);
+            
             //purge the job from the cache
             fileManager.purgeJob(id);
           } catch (Throwable e) {
@@ -249,12 +274,13 @@
    * @param jobTrackerStartTime jobtracker's start time
    * @return true if intialized properly
    *         false otherwise
-   * @deprecated Use {@link #init(JobConf, String, long, FileSystem)} instead.
+   * @deprecated Use {@link #init(JobTracker, JobConf, String, long, 
+   * FileSystem)} instead.
    */
 @Deprecated
-  public static boolean init(JobConf conf, String hostname, 
-                              long jobTrackerStartTime){
-    return init(conf, hostname, jobTrackerStartTime, null);
+  public static boolean init(JobTracker jobTracker, JobConf conf, 
+      String hostname, long jobTrackerStartTime){
+    return init(jobTracker, conf, hostname, jobTrackerStartTime, null);
   }
 
   /**
@@ -266,9 +292,8 @@
    * @return true if intialized properly
    *         false otherwise
    */
-  public static boolean init(JobConf conf, String hostname,
-                             long jobTrackerStartTime,
-                             FileSystem fs){
+  public static boolean init(JobTracker jobTracker, JobConf conf, 
+      String hostname, long jobTrackerStartTime, FileSystem fs){
     try {
       LOG_DIR = conf.get("hadoop.job.history.location" ,
         "file:///" + new File(
@@ -297,7 +322,7 @@
                      3 * 1024 * 1024);
 
       // initialize the file manager
-      fileManager = new JobHistoryFilesManager(conf);
+      fileManager = new JobHistoryFilesManager(conf, jobTracker);
     } catch(IOException e) {
         LOG.error("Failed to initialize JobHistory log file", e); 
         disableHistory = true;
@@ -1040,25 +1065,7 @@
      * This *should* be the last call to jobhistory for a given job.
      */
      static void markCompleted(JobID id) throws IOException {
-       List<Path> paths = new ArrayList<Path>();
-       Path path = fileManager.getHistoryFile(id);
-       if (path == null) {
-         LOG.info("No file for job-history with " + id + " found in cache!");
-       } else {
-         paths.add(path);
-       }
-
-       Path confPath = fileManager.getConfFileWriters(id);
-       if (confPath == null) {
-         LOG.info("No file for jobconf with " + id + " found in cache!");
-       } else {
-         paths.add(confPath);
-       }
-
-       //move the job files to done folder and purge the job
-       if (paths.size() > 0) {
-         fileManager.moveToDone(id, paths);
-       }
+       fileManager.moveToDone(id);
      }
 
      /**

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java Tue Aug 11 17:43:14 2009
@@ -300,7 +300,9 @@
     this.nonRunningReduces = new LinkedList<TaskInProgress>();    
     this.runningReduces = new LinkedHashSet<TaskInProgress>();
     this.resourceEstimator = new ResourceEstimator(this);
-    this.status = new JobStatus(jobid, 0.0f, 0.0f, JobStatus.PREP);
+    this.status = new JobStatus(jobid, 0.0f, 0.0f, JobStatus.PREP, 
+        this.profile.getUser(), this.profile.getJobName(), 
+        this.profile.getJobFile(), "");
     this.taskCompletionEvents = new ArrayList<TaskCompletionEvent>
     (numMapTasks + numReduceTasks + 10);
     
@@ -330,9 +332,9 @@
     String url = "http://" + jobtracker.getJobTrackerMachine() + ":" 
         + jobtracker.getInfoPort() + "/jobdetails.jsp?jobid=" + jobid;
     this.jobtracker = jobtracker;
-    this.status = new JobStatus(jobid, 0.0f, 0.0f, JobStatus.PREP);
+    
     this.startTime = System.currentTimeMillis();
-    status.setStartTime(startTime);
+    
     this.localFs = jobtracker.getLocalFileSystem();
 
     JobConf default_job_conf = new JobConf(default_conf);
@@ -346,10 +348,15 @@
     fs.copyToLocalFile(jobFile, localJobFile);
     conf = new JobConf(localJobFile);
     this.priority = conf.getJobPriority();
-    this.status.setJobPriority(this.priority);
     this.profile = new JobProfile(conf.getUser(), jobid, 
                                   jobFile.toString(), url, conf.getJobName(),
                                   conf.getQueueName());
+    this.status = new JobStatus(jobid, 0.0f, 0.0f, JobStatus.PREP, 
+        profile.getUser(), profile.getJobName(), profile.getJobFile(), 
+        profile.getURL().toString());
+    status.setStartTime(startTime);
+    this.status.setJobPriority(this.priority);
+
     String jarFile = conf.getJar();
     if (jarFile != null) {
       fs.copyToLocalFile(new Path(jarFile), localJarFile);
@@ -2630,6 +2637,7 @@
         this.status.setReduceProgress(1.0f);
       }
       this.finishTime = JobTracker.getClock().getTime();
+      this.status.setFinishTime(this.finishTime);
       LOG.info("Job " + this.status.getJobID() + 
                " has completed successfully.");
       
@@ -2653,11 +2661,15 @@
   private synchronized void terminateJob(int jobTerminationState) {
     if ((status.getRunState() == JobStatus.RUNNING) ||
         (status.getRunState() == JobStatus.PREP)) {
+
+      this.finishTime = JobTracker.getClock().getTime();
+      this.status.setMapProgress(1.0f);
+      this.status.setReduceProgress(1.0f);
+      this.status.setCleanupProgress(1.0f);
+      this.status.setFinishTime(this.finishTime);
+
       if (jobTerminationState == JobStatus.FAILED) {
-        this.status = new JobStatus(status.getJobID(),
-                                    1.0f, 1.0f, 1.0f, JobStatus.FAILED,
-                                    status.getJobPriority());
-        this.finishTime = JobTracker.getClock().getTime();
+        this.status.setRunState(JobStatus.FAILED);
         
         // Log the job summary
         JobSummary.logJobSummary(this, jobtracker.getClusterStatus(false));
@@ -2667,11 +2679,8 @@
                                      this.finishedMapTasks, 
                                      this.finishedReduceTasks);
       } else {
-        this.status = new JobStatus(status.getJobID(),
-                                    1.0f, 1.0f, 1.0f, JobStatus.KILLED,
-                                    status.getJobPriority());
-        this.finishTime = JobTracker.getClock().getTime();
-        
+        this.status.setRunState(JobStatus.KILLED);
+
         // Log the job summary
         JobSummary.logJobSummary(this, jobtracker.getClusterStatus(false));
 

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobStatus.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobStatus.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobStatus.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobStatus.java Tue Aug 11 17:43:14 2009
@@ -74,6 +74,14 @@
   private String user;
   private JobPriority priority;
   private String schedulingInfo="NA";
+
+  private String jobName;
+  private String jobFile;
+  private long finishTime;
+  private boolean isRetired;
+  private String historyFile = "";
+  private String trackingUrl ="";
+
     
   /**
    */
@@ -87,11 +95,17 @@
    * @param reduceProgress The progress made on the reduces
    * @param cleanupProgress The progress made on cleanup
    * @param runState The current state of the job
+   * @param user userid of the person who submitted the job.
+   * @param jobName user-specified job name.
+   * @param jobFile job configuration file. 
+   * @param trackingUrl link to the web-ui for details of the job.
    */
   public JobStatus(JobID jobid, float mapProgress, float reduceProgress,
-                   float cleanupProgress, int runState) {
+                   float cleanupProgress, int runState, 
+                   String user, String jobName, 
+                   String jobFile, String trackingUrl) {
     this(jobid, mapProgress, reduceProgress, cleanupProgress, runState,
-                  JobPriority.NORMAL);
+                  JobPriority.NORMAL, user, jobName, jobFile, trackingUrl);
   }
 
   /**
@@ -100,10 +114,16 @@
    * @param mapProgress The progress made on the maps
    * @param reduceProgress The progress made on the reduces
    * @param runState The current state of the job
+   * @param user userid of the person who submitted the job.
+   * @param jobName user-specified job name.
+   * @param jobFile job configuration file. 
+   * @param trackingUrl link to the web-ui for details of the job.
    */
   public JobStatus(JobID jobid, float mapProgress, float reduceProgress,
-                   int runState) {
-    this(jobid, mapProgress, reduceProgress, 0.0f, runState);
+                   int runState, String user, String jobName, 
+                   String jobFile, String trackingUrl) {
+    this(jobid, mapProgress, reduceProgress, 0.0f, runState, user, jobName, 
+        jobFile, trackingUrl);
   }
 
   /**
@@ -113,11 +133,17 @@
    * @param reduceProgress The progress made on the reduces
    * @param runState The current state of the job
    * @param jp Priority of the job.
+   * @param user userid of the person who submitted the job.
+   * @param jobName user-specified job name.
+   * @param jobFile job configuration file. 
+   * @param trackingUrl link to the web-ui for details of the job.
    */
    public JobStatus(JobID jobid, float mapProgress, float reduceProgress,
-                      float cleanupProgress, int runState, JobPriority jp) {
+                      float cleanupProgress, int runState, JobPriority jp, 
+                      String user, String jobName, String jobFile, 
+                      String trackingUrl) {
      this(jobid, 0.0f, mapProgress, reduceProgress, 
-          cleanupProgress, runState, jp);
+          cleanupProgress, runState, jp, user, jobName, jobFile, trackingUrl);
    }
    
   /**
@@ -129,21 +155,29 @@
    * @param cleanupProgress The progress made on the cleanup
    * @param runState The current state of the job
    * @param jp Priority of the job.
+   * @param user userid of the person who submitted the job.
+   * @param jobName user-specified job name.
+   * @param jobFile job configuration file. 
+   * @param trackingUrl link to the web-ui for details of the job.
    */
    public JobStatus(JobID jobid, float setupProgress, float mapProgress,
                     float reduceProgress, float cleanupProgress, 
-                    int runState, JobPriority jp) {
+                    int runState, JobPriority jp, String user, String jobName, 
+                    String jobFile, String trackingUrl) {
      this.jobid = jobid;
      this.setupProgress = setupProgress;
      this.mapProgress = mapProgress;
      this.reduceProgress = reduceProgress;
      this.cleanupProgress = cleanupProgress;
      this.runState = runState;
-     this.user = "nobody";
+     this.user = user;
      if (jp == null) {
        throw new IllegalArgumentException("Job Priority cannot be null.");
      }
      priority = jp;
+     this.jobName = jobName;
+     this.jobFile = jobFile;
+     this.trackingUrl = trackingUrl;
    }
    
   /**
@@ -308,6 +342,12 @@
     Text.writeString(out, user);
     WritableUtils.writeEnum(out, priority);
     Text.writeString(out, schedulingInfo);
+    out.writeLong(finishTime);
+    out.writeBoolean(isRetired);
+    Text.writeString(out, historyFile);
+    Text.writeString(out, jobName);
+    Text.writeString(out, trackingUrl);
+    Text.writeString(out, jobFile);
   }
 
   public synchronized void readFields(DataInput in) throws IOException {
@@ -321,5 +361,98 @@
     this.user = Text.readString(in);
     this.priority = WritableUtils.readEnum(in, JobPriority.class);
     this.schedulingInfo = Text.readString(in);
+    this.finishTime = in.readLong();
+    this.isRetired = in.readBoolean();
+    this.historyFile = Text.readString(in);
+    this.jobName = Text.readString(in);
+    this.trackingUrl = Text.readString(in);
+    this.jobFile = Text.readString(in);
+  }
+
+  /**
+   * Get the user-specified job name.
+   */
+  public String getJobName() {
+    return jobName;
+  }
+
+  /**
+   * Get the configuration file for the job.
+   */
+  public String getJobFile() {
+    return jobFile;
+  }
+
+  /**
+   * Get the link to the web-ui for details of the job.
+   */
+  public synchronized String getTrackingUrl() {
+    return trackingUrl;
+  }
+
+  /**
+   * Get the finish time of the job.
+   */
+  public synchronized long getFinishTime() { 
+    return finishTime;
+  }
+
+  /**
+   * Check whether the job has retired.
+   */
+  public synchronized boolean isRetired() {
+    return isRetired;
+  }
+
+  /**
+   * @return the job history file name for a completed job. If job is not 
+   * completed or history file not available then return null.
+   */
+  public synchronized String getHistoryFile() {
+    return historyFile;
+  }
+
+ /** 
+   * Set the finish time of the job
+   * @param finishTime The finishTime of the job
+   */
+  synchronized void setFinishTime(long finishTime) {
+    this.finishTime = finishTime;
+  }
+
+  /**
+   * Set the job history file url for a completed job
+   */
+  synchronized void setHistoryFile(String historyFile) {
+    this.historyFile = historyFile;
+  }
+
+  /**
+   * Set the link to the web-ui for details of the job.
+   */
+  synchronized void setTrackingUrl(String trackingUrl) {
+    this.trackingUrl = trackingUrl;
+  }
+
+  /**
+   * Set the job retire flag to true.
+   */
+  synchronized void setRetired() {
+    this.isRetired = true;
+  }
+
+  public String toString() {
+    StringBuffer buffer = new StringBuffer();
+    buffer.append("job-id : " + jobid);
+    buffer.append("map-progress : " + mapProgress);
+    buffer.append("reduce-progress : " + reduceProgress);
+    buffer.append("cleanup-progress : " + cleanupProgress);
+    buffer.append("setup-progress : " + setupProgress);
+    buffer.append("runstate : " + runState);
+    buffer.append("start-time : " + startTime);
+    buffer.append("user-name : " + user);
+    buffer.append("priority : " + priority);
+    buffer.append("scheduling-info : " + schedulingInfo);
+    return buffer.toString();
   }
 }

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobSubmissionProtocol.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobSubmissionProtocol.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobSubmissionProtocol.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobSubmissionProtocol.java Tue Aug 11 17:43:14 2009
@@ -65,8 +65,9 @@
    *             Part of HADOOP-5913.  
    * Version 24: Modified ClusterStatus to include BlackListInfo class which 
    *             encapsulates reasons and report for blacklisted node.          
+   * Version 25: Added fields to JobStatus for HADOOP-817.
    */
-  public static final long versionID = 24L;
+  public static final long versionID = 25L;
 
   /**
    * Allocate a name for the job.

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java Tue Aug 11 17:43:14 2009
@@ -19,12 +19,15 @@
 
 
 import java.io.IOException;
+import java.io.UnsupportedEncodingException;
 import java.net.BindException;
 import java.net.InetSocketAddress;
+import java.net.URLEncoder;
 import java.net.UnknownHostException;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
@@ -33,6 +36,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
@@ -417,14 +421,56 @@
     }
   }
 
+  synchronized void historyFileCopied(JobID jobid, String historyFile) {
+    JobStatus status = getJobStatus(jobid);
+    if (status != null) {
+      String trackingUrl = "";
+      if (historyFile != null) {
+        status.setHistoryFile(historyFile);
+        try {
+          trackingUrl = "http://" + getJobTrackerMachine() + ":" + 
+            getInfoPort() + "/jobdetailshistory.jsp?jobid=" + 
+            jobid + "&logFile=" + URLEncoder.encode(historyFile, "UTF-8");
+        } catch (UnsupportedEncodingException e) {
+          LOG.warn("Could not create trackingUrl", e);
+        }
+      }
+      status.setTrackingUrl(trackingUrl);
+    }
+  }
+
   ///////////////////////////////////////////////////////
   // Used to remove old finished Jobs that have been around for too long
   ///////////////////////////////////////////////////////
   class RetireJobs implements Runnable {
     int runCount = 0;
+    private final Map<JobID, JobStatus> jobIDStatusMap = 
+      new HashMap<JobID, JobStatus>();
+    private final LinkedList<JobStatus> jobStatusQ = 
+      new LinkedList<JobStatus>();
     public RetireJobs() {
     }
 
+    synchronized void addToCache(JobStatus status) {
+      status.setRetired();
+      jobStatusQ.add(status);
+      jobIDStatusMap.put(status.getJobID(), status);
+      if (jobStatusQ.size() > retiredJobsCacheSize) {
+        JobStatus removed = jobStatusQ.remove();
+        jobIDStatusMap.remove(removed.getJobID());
+        LOG.info("Retired job removed from cache " + removed.getJobID());
+      }
+    }
+
+    synchronized JobStatus get(JobID jobId) {
+      return jobIDStatusMap.get(jobId);
+    }
+
+    @SuppressWarnings("unchecked")
+    synchronized LinkedList<JobStatus> getAll() {
+      return (LinkedList<JobStatus>) jobStatusQ.clone();
+    }
+
     /**
      * The run method lives for the life of the JobTracker,
      * and removes Jobs that are not still running, but which
@@ -476,6 +522,7 @@
 
                     // clean up job files from the local disk
                     JobHistory.JobInfo.cleanupJob(job.getProfile().getJobID());
+                    addToCache(job.getStatus());
                   }
                 }
               }
@@ -1723,6 +1770,7 @@
   Thread expireTrackersThread = null;
   RetireJobs retireJobs = new RetireJobs();
   Thread retireJobsThread = null;
+  final int retiredJobsCacheSize;
   ExpireLaunchingTasks expireLaunchingTasks = new ExpireLaunchingTasks();
   Thread expireLaunchingTaskThread = new Thread(expireLaunchingTasks,
                                                 "expireLaunchingTasks");
@@ -1802,6 +1850,8 @@
       conf.getLong("mapred.tasktracker.expiry.interval", 10 * 60 * 1000);
     retireJobInterval = conf.getLong("mapred.jobtracker.retirejob.interval", 24 * 60 * 60 * 1000);
     retireJobCheckInterval = conf.getLong("mapred.jobtracker.retirejob.check", 60 * 1000);
+    retiredJobsCacheSize = 
+      conf.getInt("mapred.job.tracker.retiredjobs.cache.size", 1000);
     // min time before retire
     MIN_TIME_BEFORE_RETIRE = conf.getInt("mapred.jobtracker.retirejob.interval.min", 60000);
     MAX_COMPLETE_USER_JOBS_IN_MEMORY = conf.getInt("mapred.jobtracker.completeuserjobs.maximum", 100);
@@ -1872,7 +1922,7 @@
         tmpInfoPort == 0, conf);
     infoServer.setAttribute("job.tracker", this);
     // initialize history parameters.
-    boolean historyInitialized = JobHistory.init(conf, this.localMachine,
+    boolean historyInitialized = JobHistory.init(this, conf, this.localMachine,
                                                  this.startTime);
     
     infoServer.addServlet("reducegraph", "/taskgraph", TaskGraphServlet.class);
@@ -2420,12 +2470,6 @@
             while (userJobs.size() > 
                    MAX_COMPLETE_USER_JOBS_IN_MEMORY) {
               JobInProgress rjob = userJobs.get(0);
-                
-              // Do not delete 'current'
-              // finished job just yet.
-              if (rjob == job) {
-                break;
-              }
 
               // do not retire jobs that finished in the very recent past.
               if (rjob.getFinishTime() + MIN_TIME_BEFORE_RETIRE > now) {
@@ -2455,6 +2499,9 @@
                 LOG.info("Retired job with id: '" + 
                          rjob.getProfile().getJobID() + "' of user: '" +
                          jobUser + "'");
+                // clean up job files from the local disk
+                JobHistory.JobInfo.cleanupJob(rjob.getProfile().getJobID());
+                retireJobs.addToCache(rjob.getStatus());
               } else {
                 // Do not remove jobs that aren't complete.
                 // Stop here, and let the next pass take
@@ -3657,7 +3704,13 @@
       JobInProgress job = jobs.get(jobid);
       if (job != null) {
         return job.getStatus();
-      } 
+      } else {
+        
+        JobStatus status = retireJobs.get(jobid);
+        if (status != null) {
+          return status;
+        }
+      }
     }
     return completedJobStatusStore.readJobStatus(jobid);
   }
@@ -3796,19 +3849,19 @@
    */
   public synchronized String[] getTaskDiagnostics(TaskAttemptID taskId)  
     throws IOException {
-    
+    List<String> taskDiagnosticInfo = null;
     JobID jobId = taskId.getJobID();
     TaskID tipId = taskId.getTaskID();
     JobInProgress job = jobs.get(jobId);
-    if (job == null) {
-      throw new IllegalArgumentException("Job " + jobId + " not found.");
-    }
-    TaskInProgress tip = job.getTaskInProgress(tipId);
-    if (tip == null) {
-      throw new IllegalArgumentException("TIP " + tipId + " not found.");
+    if (job != null) {
+      TaskInProgress tip = job.getTaskInProgress(tipId);
+      if (tip != null) {
+        taskDiagnosticInfo = tip.getDiagnosticInfo(taskId);
+      }
+      
     }
-    List<String> taskDiagnosticInfo = tip.getDiagnosticInfo(taskId);
-    return ((taskDiagnosticInfo == null) ? null 
+    
+    return ((taskDiagnosticInfo == null) ? new String[0] 
             : taskDiagnosticInfo.toArray(new String[0]));
   }
     
@@ -3877,7 +3930,10 @@
   } 
   
   public JobStatus[] getAllJobs() {
-    return getJobStatus(jobs.values(),false);
+    List<JobStatus> list = new ArrayList<JobStatus>();
+    list.addAll(Arrays.asList(getJobStatus(jobs.values(),false)));
+    list.addAll(retireJobs.getAll());
+    return list.toArray(new JobStatus[list.size()]);
   }
     
   /**

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java Tue Aug 11 17:43:14 2009
@@ -120,7 +120,9 @@
       this.job = new JobConf(localFile);
       profile = new JobProfile(job.getUser(), id, file.toString(), 
                                "http://localhost:8080/", job.getJobName());
-      status = new JobStatus(id, 0.0f, 0.0f, JobStatus.RUNNING);
+      status = new JobStatus(id, 0.0f, 0.0f, JobStatus.RUNNING, 
+          profile.getUser(), profile.getJobName(), profile.getJobFile(), 
+          profile.getURL().toString());
 
       jobs.put(id, this);
 

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/RunningJob.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/RunningJob.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/RunningJob.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/RunningJob.java Tue Aug 11 17:43:14 2009
@@ -189,4 +189,22 @@
    * @throws IOException
    */
   public String[] getTaskDiagnostics(TaskAttemptID taskid) throws IOException;
+
+  /**
+   * Get the url where history file is archived. Returns empty string if 
+   * history file is not available yet. 
+   * 
+   * @return the url where history file is archived
+   * @throws IOException
+   */
+  public String getHistoryUrl() throws IOException;
+
+  /**
+   * Check whether the job has been removed from JobTracker memory and retired.
+   * On retire, the job history file is copied to a location known by 
+   * {@link #getHistoryUrl()}
+   * @return <code>true</code> if the job retired, else <code>false</code>.
+   * @throws IOException
+   */
+  public boolean isRetired() throws IOException;
 }

Modified: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobHistoryParsing.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobHistoryParsing.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobHistoryParsing.java (original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobHistoryParsing.java Tue Aug 11 17:43:14 2009
@@ -59,7 +59,7 @@
     JobConf conf = new JobConf();
     conf.set("hadoop.job.history.location", historyDir.toString());
     FileSystem fs = FileSystem.getLocal(new JobConf());
-    JobHistory.init(conf, "localhost", 1234, fs);
+    JobHistory.init(null, conf, "localhost", 1234, fs);
     Path historyLog = new Path(historyDir, "testlog");
     PrintWriter out = new PrintWriter(fs.create(historyLog));
     historyWriter.add(out);

Modified: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobQueueTaskScheduler.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobQueueTaskScheduler.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobQueueTaskScheduler.java (original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobQueueTaskScheduler.java Tue Aug 11 17:43:14 2009
@@ -50,7 +50,9 @@
       super(new JobID("test", ++jobCounter), jobConf, null);
       this.taskTrackerManager = taskTrackerManager;
       this.startTime = System.currentTimeMillis();
-      this.status = new JobStatus(getJobID(), 0f, 0f, JobStatus.PREP);
+      this.status = new JobStatus(getJobID(), 0f, 0f, JobStatus.PREP, 
+          jobConf.getUser(), 
+          jobConf.getJobName(), "", "");
       this.status.setJobPriority(JobPriority.NORMAL);
       this.status.setStartTime(startTime);
     }

Modified: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java (original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestJobRetire.java Tue Aug 11 17:43:14 2009
@@ -18,9 +18,10 @@
 
 package org.apache.hadoop.mapred;
 
+import java.io.File;
+import java.io.IOException;
+
 import junit.framework.TestCase;
-import org.apache.hadoop.hdfs.MiniDFSCluster;
-import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.mapred.UtilsForTests.FakeClock;
 
@@ -36,26 +37,31 @@
              "job-expiry-testing");
   private static final Log LOG = LogFactory.getLog(TestJobRetire.class);
 
+  private void testJobConfFile(JobID id, boolean exists) {
+    // get the job conf filename
+    String name = JobHistory.JobInfo.getLocalJobFilePath(id);
+    File file = new File(name);
+ 
+    assertEquals("JobConf file check failed", exists, file.exists());
+  }
+
   /** Test if the job after completion waits for atleast 
    *  mapred.jobtracker.retirejob.interval.min amount of time.
    */
   public void testMinIntervalBeforeRetire() throws Exception {
-    MiniDFSCluster dfs = null;
     MiniMRCluster mr = null;
     int min = 10000;
     int max = 5000;
     try {
       FakeClock clock = new FakeClock();
       JobConf conf = new JobConf();
-      dfs = new MiniDFSCluster(conf, 1, true, null);
-      FileSystem fileSys = dfs.getFileSystem();
-      String namenode = fileSys.getUri().toString();
 
       conf.setLong("mapred.jobtracker.retirejob.check", 1000); // 1 sec
       conf.setInt("mapred.jobtracker.retirejob.interval.min", min); //10 secs
       conf.setInt("mapred.jobtracker.retirejob.interval", max); // 5 secs
       conf.setBoolean("mapred.job.tracker.persist.jobstatus.active", false);
-      mr = new MiniMRCluster(0, 0, 1, namenode, 1, null, null, null, conf, 0, 
+      conf.setInt("mapred.job.tracker.retiredjobs.cache.size", 0);
+      mr = new MiniMRCluster(0, 0, 1, "file:///", 1, null, null, null, conf, 0, 
                              clock);
       JobConf jobConf = mr.createJobConf();
       JobTracker jobtracker = mr.getJobTrackerRunner().getJobTracker();
@@ -82,6 +88,9 @@
 
       assertNotNull(jc.getJob(id));
 
+      //check that the job is not retired
+      assertFalse(jc.getJob(id).isRetired());
+      
       // snapshot expiry thread count
       snapshot = jobtracker.retireJobs.runCount;
       clock.advance(min - max); // adv to expiry min time
@@ -93,9 +102,11 @@
 
       // check if the job is missing
       assertNull(jc.getJob(id));
+      assertNull(jobtracker.getJob(id));
+
+      testJobConfFile(id, false);
     } finally {
       if (mr != null) { mr.shutdown();}
-      if (dfs != null) { dfs.shutdown();}
     }
   }
 
@@ -103,22 +114,19 @@
    *  mapred.jobtracker.retirejob.interval amount after the time.
    */
   public void testJobRetire() throws Exception {
-    MiniDFSCluster dfs = null;
     MiniMRCluster mr = null;
     int min = 10000;
     int max = 20000;
     try {
       FakeClock clock = new FakeClock();
       JobConf conf = new JobConf();
-      dfs = new MiniDFSCluster(conf, 1, true, null);
-      FileSystem fileSys = dfs.getFileSystem();
-      String namenode = fileSys.getUri().toString();
 
       conf.setLong("mapred.jobtracker.retirejob.check", 1000); // 1 sec
       conf.setInt("mapred.jobtracker.retirejob.interval.min", min); // 10 secs
       conf.setInt("mapred.jobtracker.retirejob.interval", max); // 20 secs
       conf.setBoolean("mapred.job.tracker.persist.jobstatus.active", false);
-      mr = new MiniMRCluster(0, 0, 1, namenode, 1, null, null, null, conf, 0, 
+      conf.setInt("mapred.job.tracker.retiredjobs.cache.size", 0);
+      mr = new MiniMRCluster(0, 0, 1, "file:///", 1, null, null, null, conf, 0, 
                              clock);
       JobConf jobConf = mr.createJobConf();
       JobTracker jobtracker = mr.getJobTrackerRunner().getJobTracker();
@@ -141,12 +149,169 @@
         // wait for the thread to run
         UtilsForTests.waitFor(1000);
       }
-
+ 
       // check if the job is missing
       assertNull(jc.getJob(id));
+      assertNull(jobtracker.getJob(id));
+
+      testJobConfFile(id, false);
     } finally {
       if (mr != null) { mr.shutdown();}
-      if (dfs != null) { dfs.shutdown();}
     }
   }
+
+  /** Test if the job after gets expired after
+   *  mapred.jobtracker.completeuserjobs.maximum jobs.
+   */
+  public void testMaxJobRetire() throws Exception {
+    MiniMRCluster mr = null;
+    int min = 10000;
+    int max = 20000;
+    try {
+      FakeClock clock = new FakeClock();
+      JobConf conf = new JobConf();
+      
+      conf.setLong("mapred.jobtracker.retirejob.check", 1000); // 1 sec
+      conf.setInt("mapred.jobtracker.retirejob.interval.min", min); // 10 secs
+      conf.setInt("mapred.jobtracker.retirejob.interval", max); // 20 secs
+      conf.setBoolean("mapred.job.tracker.persist.jobstatus.active", false);
+      conf.setInt("mapred.jobtracker.completeuserjobs.maximum", 1);
+      conf.setInt("mapred.job.tracker.retiredjobs.cache.size", 0);
+      mr = new MiniMRCluster(0, 0, 1, "file:///", 1, null, null, null, conf, 0, 
+                             clock);
+      JobConf jobConf = mr.createJobConf();
+      JobTracker jobtracker = mr.getJobTrackerRunner().getJobTracker();
+      
+      Path inDir = new Path(testDir, "input2.1");
+      Path outDir = new Path(testDir, "output2.1");
+      RunningJob rj = UtilsForTests.runJob(jobConf, inDir, outDir, 0, 0);
+      rj.waitForCompletion();
+      JobID id = rj.getID();
+      JobClient jc = new JobClient(jobConf);
+
+      // check if the job is successful
+      assertTrue(rj.isSuccessful());
+
+      clock.advance(min + 1); // adv to expiry min time
+
+      inDir = new Path(testDir, "input2.2");
+      outDir = new Path(testDir, "output2.2");
+      RunningJob rj2 = UtilsForTests.runJob(jobConf, inDir, outDir, 0, 0);
+      rj2.waitForCompletion();
+      JobID id2 = rj2.getID();
+
+      // check if the job#1 is missing
+      assertNull(jc.getJob(id));
+      assertNull("Job still not missing from jobtracker", jobtracker.getJob(id));
+      
+      // check if the job#2 exists
+      assertNotNull(jc.getJob(id2));
+      assertNotNull("Job " + id2 + " missing at the jobtracker before expiry", 
+                    jobtracker.getJob(id2));
+
+      testJobConfFile(id, false);
+      testJobConfFile(id2, true);
+    } finally {
+      if (mr != null) {mr.shutdown();}
+    }
+  }
+
+  /** Test if the job after gets expired but basic info is cached with jobtracker
+   */
+  public void testRetiredJobCache() throws Exception {
+    MiniMRCluster mr = null;
+    int min = 10000;
+    int max = 20000;
+    try {
+      FakeClock clock = new FakeClock();
+      JobConf conf = new JobConf();
+
+      conf.setLong("mapred.jobtracker.retirejob.check", 1000); // 1 sec
+      conf.setInt("mapred.jobtracker.retirejob.interval.min", min); // 10 secs
+      conf.setInt("mapred.jobtracker.retirejob.interval", max); // 20 secs
+      conf.setBoolean("mapred.job.tracker.persist.jobstatus.active", false);
+      conf.setInt("mapred.jobtracker.completeuserjobs.maximum", 1);
+      conf.setInt("mapred.job.tracker.retiredjobs.cache.size", 1);
+      mr = new MiniMRCluster(0, 0, 1, "file:///", 1, null, null, null, conf, 0,
+                             clock);
+      JobConf jobConf = mr.createJobConf();
+      JobTracker jobtracker = mr.getJobTrackerRunner().getJobTracker();
+
+      Path inDir = new Path(testDir, "input3.1");
+      Path outDir = new Path(testDir, "output3.1");
+      RunningJob rj = UtilsForTests.runJob(jobConf, inDir, outDir, 0, 0);
+      rj.waitForCompletion();
+      JobID id = rj.getID();
+      JobClient jc = new JobClient(jobConf);
+
+      // check if the job is successful
+      assertTrue(rj.isSuccessful());
+      JobStatus status1 = jobtracker.getJobStatus(id);
+
+      clock.advance(min + 1); // adv to expiry min time
+
+      inDir = new Path(testDir, "input3.2");
+      outDir = new Path(testDir, "output3.2");
+      RunningJob rj2 = UtilsForTests.runJob(jobConf, inDir, outDir, 0, 0);
+      rj2.waitForCompletion();
+      JobID id2 = rj2.getID();
+      JobStatus status2 = jobtracker.getJobStatus(id2);
+
+      // check if the job#1 is missing in jt but cached status
+      assertNotNull("Job status missing from status cache", jc.getJob(id));
+      // check the status at jobtracker
+      assertEquals("Status mismatch for job " + id, status1.toString(), 
+                   jobtracker.getJobStatus(id).toString());
+      testRetiredCachedJobStatus(status1, rj);
+      assertNull("Job still not missing from jobtracker", jobtracker.getJob(id));
+
+      // check if the job#2 exists
+      assertNotNull(jc.getJob(id2));
+      // check the status .. 
+      
+      assertNotNull("Job " + id2 + " missing at the jobtracker before expiry",
+                    jobtracker.getJob(id2));
+
+      testJobConfFile(id, false);
+      testJobConfFile(id2, true);
+
+      clock.advance(min + 1); // adv to expiry min time
+
+      inDir = new Path(testDir, "input3.3");
+      outDir = new Path(testDir, "output3.3");
+      RunningJob rj3 = UtilsForTests.runJob(jobConf, inDir, outDir, 0, 0);
+      rj3.waitForCompletion();
+      JobID id3 = rj3.getID();
+
+      // check if the job#1 is missing in all the caches
+      assertNull("Job status still in status cache", jc.getJob(id));
+      // check if the job#2 is missing in jt but cached status
+      assertNotNull(jc.getJob(id2));
+      assertEquals("Status mismatch for job " + id2, status2.toString(), 
+                   jobtracker.getJobStatus(id2).toString());
+      testRetiredCachedJobStatus(status2, rj2);
+      assertNull("Job " + id2 + " missing at the jobtracker before expiry",
+                 jobtracker.getJob(id2));
+      // check if the job#3 exists
+      assertNotNull(jc.getJob(id3));
+      assertNotNull("Job " + id3 + " missing at the jobtracker before expiry",
+                    jobtracker.getJob(id3));
+    } finally {
+      if (mr != null) {mr.shutdown();}
+    }
+  }
+
+  private static void testRetiredCachedJobStatus(JobStatus status, 
+                                                 RunningJob rj) 
+  throws IOException {
+    assertEquals(status.getJobID(), rj.getID());
+    assertEquals(status.mapProgress(), rj.mapProgress());
+    assertEquals(status.reduceProgress(), rj.reduceProgress());
+    assertEquals(status.setupProgress(), rj.setupProgress());
+    assertEquals(status.cleanupProgress(), rj.cleanupProgress());
+    assertEquals(status.getRunState(), rj.getJobState());
+    assertEquals(status.getJobName(), rj.getJobName());
+    assertEquals(status.getTrackingUrl(), rj.getTrackingURL());
+    assertEquals(status.isRetired(), true);
+  }
 }

Modified: hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestParallelInitialization.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestParallelInitialization.java?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestParallelInitialization.java (original)
+++ hadoop/mapreduce/trunk/src/test/mapred/org/apache/hadoop/mapred/TestParallelInitialization.java Tue Aug 11 17:43:14 2009
@@ -46,7 +46,9 @@
         FakeTaskTrackerManager taskTrackerManager) throws IOException {
       super(new JobID("test", ++jobCounter), jobConf, null);
       this.startTime = System.currentTimeMillis();
-      this.status = new JobStatus(getJobID(), 0f, 0f, JobStatus.PREP);
+      this.status = new JobStatus(getJobID(), 0f, 0f, JobStatus.PREP, 
+          jobConf.getUser(), 
+          jobConf.getJobName(), "", "");
       this.status.setJobPriority(JobPriority.NORMAL);
       this.status.setStartTime(startTime);
     }

Modified: hadoop/mapreduce/trunk/src/webapps/job/analysejobhistory.jsp
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/webapps/job/analysejobhistory.jsp?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/webapps/job/analysejobhistory.jsp (original)
+++ hadoop/mapreduce/trunk/src/webapps/job/analysejobhistory.jsp Tue Aug 11 17:43:14 2009
@@ -23,14 +23,12 @@
   import="java.io.*"
   import="java.util.*"
   import="org.apache.hadoop.mapred.*"
+  import="org.apache.hadoop.fs.*"
   import="org.apache.hadoop.util.*"
   import="java.text.SimpleDateFormat"
   import="org.apache.hadoop.mapred.JobHistory.*"
 %>
-<jsp:include page="loadhistory.jsp">
-  <jsp:param name="jobid" value="<%=request.getParameter("jobid") %>"/>
-  <jsp:param name="logFile" value="<%=request.getParameter("logFile") %>"/>
-</jsp:include>
+
 <%!	private static SimpleDateFormat dateFormat 
                               = new SimpleDateFormat("d/MM HH:mm:ss") ; 
 %>
@@ -46,7 +44,8 @@
   if (numTasks != null) {
     showTasks = Integer.parseInt(numTasks);  
   }
-  JobInfo job = (JobInfo)request.getSession().getAttribute("job");
+  FileSystem fs = (FileSystem) application.getAttribute("fileSys");
+  JobInfo job = JSPUtil.getJobInfo(request, fs);
 %>
 <h2>Hadoop Job <a href="jobdetailshistory.jsp?jobid=<%=jobid%>&&logFile=<%=encodedLogFileName%>"><%=jobid %> </a></h2>
 <b>User : </b> <%=job.get(Keys.USER) %><br/> 

Modified: hadoop/mapreduce/trunk/src/webapps/job/jobconf_history.jsp
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/webapps/job/jobconf_history.jsp?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/webapps/job/jobconf_history.jsp (original)
+++ hadoop/mapreduce/trunk/src/webapps/job/jobconf_history.jsp Tue Aug 11 17:43:14 2009
@@ -51,7 +51,7 @@
   Path logDir = new Path(request.getParameter("jobLogDir"));
   Path jobFilePath = new Path(logDir, 
                        request.getParameter("jobUniqueString") + "_conf.xml");
-  FileSystem fs = (FileSystem)request.getSession().getAttribute("fs");
+  FileSystem fs = (FileSystem) application.getAttribute("fileSys");
   FSDataInputStream jobFile = null; 
   try {
     jobFile = fs.open(jobFilePath);

Modified: hadoop/mapreduce/trunk/src/webapps/job/jobdetailshistory.jsp
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/webapps/job/jobdetailshistory.jsp?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/webapps/job/jobdetailshistory.jsp (original)
+++ hadoop/mapreduce/trunk/src/webapps/job/jobdetailshistory.jsp Tue Aug 11 17:43:14 2009
@@ -30,10 +30,7 @@
 %>
 <%!	private static final long serialVersionUID = 1L;
 %>
-<jsp:include page="loadhistory.jsp">
-  <jsp:param name="jobid" value="<%=request.getParameter("jobid") %>"/>
-  <jsp:param name="logFile" value="<%=request.getParameter("logFile") %>"/>
-</jsp:include>
+
 <%! static SimpleDateFormat dateFormat = new SimpleDateFormat("d-MMM-yyyy HH:mm:ss") ; %>
 <%
     String jobid = request.getParameter("jobid");
@@ -44,8 +41,8 @@
     String[] jobDetails = jobFile.getName().split("_");
     String jobUniqueString = jobDetails[0] + "_" +jobDetails[1] + "_" + jobid ;
 	
-    JobInfo job = (JobInfo)request.getSession().getAttribute("job");
-    FileSystem fs = (FileSystem)request.getSession().getAttribute("fs");
+    FileSystem fs = (FileSystem) application.getAttribute("fileSys");
+    JobInfo job = JSPUtil.getJobInfo(request, fs);
 %>
 <html><body>
 <h2>Hadoop Job <%=jobid %> on <a href="jobhistory.jsp">History Viewer</a></h2>

Modified: hadoop/mapreduce/trunk/src/webapps/job/jobtaskshistory.jsp
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/webapps/job/jobtaskshistory.jsp?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/webapps/job/jobtaskshistory.jsp (original)
+++ hadoop/mapreduce/trunk/src/webapps/job/jobtaskshistory.jsp Tue Aug 11 17:43:14 2009
@@ -23,14 +23,12 @@
   import="java.io.*"
   import="java.util.*"
   import="org.apache.hadoop.mapred.*"
+  import="org.apache.hadoop.fs.*"
   import="org.apache.hadoop.util.*"
   import="java.text.SimpleDateFormat"
   import="org.apache.hadoop.mapred.JobHistory.*"
 %>
-<jsp:include page="loadhistory.jsp">
-	<jsp:param name="jobid" value="<%=request.getParameter("jobid") %>"/>
-	<jsp:param name="logFile" value="<%=request.getParameter("logFile") %>"/>
-</jsp:include>
+
 <%!	
   private static SimpleDateFormat dateFormat =
                                     new SimpleDateFormat("d/MM HH:mm:ss") ; 
@@ -45,8 +43,8 @@
   String taskStatus = request.getParameter("status"); 
   String taskType = request.getParameter("taskType"); 
   
-  JobHistory.JobInfo job = (JobHistory.JobInfo)request.
-                            getSession().getAttribute("job");
+  FileSystem fs = (FileSystem) application.getAttribute("fileSys");
+  JobInfo job = JSPUtil.getJobInfo(request, fs);
   Map<String, JobHistory.Task> tasks = job.getAllTasks(); 
 %>
 <html>

Modified: hadoop/mapreduce/trunk/src/webapps/job/jobtracker.jsp
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/webapps/job/jobtracker.jsp?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/webapps/job/jobtracker.jsp (original)
+++ hadoop/mapreduce/trunk/src/webapps/job/jobtracker.jsp Tue Aug 11 17:43:14 2009
@@ -100,8 +100,7 @@
   <ul id="quicklinks-list">
     <li><a href="#scheduling_info">Scheduling Info</a></li>
     <li><a href="#running_jobs">Running Jobs</a></li>
-    <li><a href="#completed_jobs">Completed Jobs</a></li>
-    <li><a href="#failed_jobs">Failed Jobs</a></li>
+    <li><a href="#retired_jobs">Retired Jobs</a></li>
     <li><a href="#local_logs">Local Logs</a></li>
   </ul>
 </div>
@@ -160,13 +159,27 @@
 <%=JSPUtil.generateJobTable("Running", runningJobs, 30, 0)%>
 <hr>
 
-<h2 id="completed_jobs">Completed Jobs</h2>
-<%=JSPUtil.generateJobTable("Completed", completedJobs, 0, runningJobs.size())%>
-<hr>
+<%
+if (completedJobs.size() > 0) {
+  out.print("<h2 id=\"completed_jobs\">Completed Jobs</h2>");
+  out.print(JSPUtil.generateJobTable("Completed", completedJobs, 0, 
+    runningJobs.size()));
+  out.print("<hr>");
+}
+%>
+
+<%
+if (failedJobs.size() > 0) {
+  out.print("<h2 id=\"failed_jobs\">Failed Jobs</h2>");
+  out.print(JSPUtil.generateJobTable("Failed", failedJobs, 0, 
+    (runningJobs.size()+completedJobs.size())));
+  out.print("<hr>");
+}
+%>
 
-<h2 id="failed_jobs">Failed Jobs</h2>
-<%=JSPUtil.generateJobTable("Failed", failedJobs, 0, 
-    (runningJobs.size()+completedJobs.size()))%>
+<h2 id="retired_jobs">Retired Jobs</h2>
+<%=JSPUtil.generateRetiredJobTable(tracker, 
+  (runningJobs.size()+completedJobs.size()+failedJobs.size()))%>
 <hr>
 
 <h2 id="local_logs">Local Logs</h2>

Modified: hadoop/mapreduce/trunk/src/webapps/job/taskdetailshistory.jsp
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/webapps/job/taskdetailshistory.jsp?rev=803231&r1=803230&r2=803231&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/webapps/job/taskdetailshistory.jsp (original)
+++ hadoop/mapreduce/trunk/src/webapps/job/taskdetailshistory.jsp Tue Aug 11 17:43:14 2009
@@ -23,14 +23,12 @@
   import="java.io.*"
   import="java.util.*"
   import="org.apache.hadoop.mapred.*"
+  import="org.apache.hadoop.fs.*"
   import="org.apache.hadoop.util.*"
   import="java.text.SimpleDateFormat"
   import="org.apache.hadoop.mapred.JobHistory.*"
 %>
-<jsp:include page="loadhistory.jsp">
-  <jsp:param name="jobid" value="<%=request.getParameter("jobid") %>"/>
-  <jsp:param name="jobTrackerId" value="<%=request.getParameter("jobTrackerId") %>"/>
-</jsp:include>
+
 <%!	private static SimpleDateFormat dateFormat = new SimpleDateFormat("d/MM HH:mm:ss") ; %>
 <%!	private static final long serialVersionUID = 1L;
 %>
@@ -40,8 +38,8 @@
   String logFile = request.getParameter("logFile");
   String encodedLogFileName = JobHistory.JobInfo.encodeJobHistoryFilePath(logFile);
   String taskid = request.getParameter("taskid"); 
-  JobHistory.JobInfo job = (JobHistory.JobInfo)
-                              request.getSession().getAttribute("job");
+  FileSystem fs = (FileSystem) application.getAttribute("fileSys");
+  JobInfo job = JSPUtil.getJobInfo(request, fs);
   JobHistory.Task task = job.getAllTasks().get(taskid); 
   String type = task.get(Keys.TASK_TYPE);
 %>



Mime
View raw message