hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From szets...@apache.org
Subject svn commit: r1466658 - in /hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project: ./ conf/ hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/ hadoop-mapreduce-client/hadoop-mapreduce-clien...
Date Wed, 10 Apr 2013 20:17:56 GMT
Author: szetszwo
Date: Wed Apr 10 20:17:39 2013
New Revision: 1466658

URL: http://svn.apache.org/r1466658
Log:
Merge r1464808 through r1466652 from trunk.

Added:
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryServer.java
      - copied unchanged from r1466652, hadoop/common/trunk/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryServer.java
Modified:
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/   (props changed)
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/CHANGES.txt   (contents, props changed)
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/conf/   (props changed)
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/avro/Events.avpr
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobSubmittedEvent.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml   (props changed)
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryFileManager.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestCompletedTask.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryEntities.java
    hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryParsing.java

Propchange: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/
------------------------------------------------------------------------------
  Merged /hadoop/common/trunk/hadoop-mapreduce-project:r1464808-1466652

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/CHANGES.txt?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/CHANGES.txt (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/CHANGES.txt Wed Apr 10 20:17:39 2013
@@ -166,6 +166,9 @@ Release 2.0.5-alpha - UNRELEASED
 
   IMPROVEMENTS
 
+    MAPREDUCE-5129. Allow tags to JobHistory for deeper analytics. (billie via
+    acmurthy)
+
   OPTIMIZATIONS
 
   BUG FIXES
@@ -222,6 +225,9 @@ Release 2.0.4-beta - UNRELEASED
     MAPREDUCE-5088. MR Client gets an renewer token exception while Oozie is
     submitting a job (Daryn Sharp via cos)
 
+    MAPREDUCE-5138. Fix LocalDistributedCacheManager after YARN-112. (Omkar Vinit
+    Joshi via vinodkv)
+
 Release 2.0.3-alpha - 2013-02-06 
 
   INCOMPATIBLE CHANGES
@@ -773,6 +779,9 @@ Release 0.23.7 - UNRELEASED
 
     MAPREDUCE-4991. coverage for gridmix (Aleksey Gorshkov via tgraves)
 
+    MAPREDUCE-5007. fix coverage org.apache.hadoop.mapreduce.v2.hs (Aleksey 
+    Gorshkov via tgraves)
+
 Release 0.23.6 - UNRELEASED
 
   INCOMPATIBLE CHANGES

Propchange: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/CHANGES.txt
------------------------------------------------------------------------------
  Merged /hadoop/common/trunk/hadoop-mapreduce-project/CHANGES.txt:r1464808-1466652

Propchange: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/conf/
------------------------------------------------------------------------------
  Merged /hadoop/common/trunk/hadoop-mapreduce-project/conf:r1464808-1466652

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java Wed Apr 10 20:17:39 2013
@@ -1305,7 +1305,8 @@ public class JobImpl implements org.apac
             job.conf.get(MRJobConfig.WORKFLOW_ID, ""),
             job.conf.get(MRJobConfig.WORKFLOW_NAME, ""),
             job.conf.get(MRJobConfig.WORKFLOW_NODE_NAME, ""),
-            getWorkflowAdjacencies(job.conf));
+            getWorkflowAdjacencies(job.conf),
+            job.conf.get(MRJobConfig.WORKFLOW_TAGS, ""));
         job.eventHandler.handle(new JobHistoryEvent(job.jobId, jse));
         //TODO JH Verify jobACLs, UserName via UGI?
 

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestJobImpl.java Wed Apr 10 20:17:39 2013
@@ -114,6 +114,7 @@ public class TestJobImpl {
     conf.set(MRJobConfig.WORKFLOW_NODE_NAME, "testNodeName");
     conf.set(MRJobConfig.WORKFLOW_ADJACENCY_PREFIX_STRING + "key1", "value1");
     conf.set(MRJobConfig.WORKFLOW_ADJACENCY_PREFIX_STRING + "key2", "value2");
+    conf.set(MRJobConfig.WORKFLOW_TAGS, "tag1,tag2");
     
  
     AsyncDispatcher dispatcher = new AsyncDispatcher();
@@ -126,7 +127,8 @@ public class TestJobImpl {
     commitHandler.start();
 
     JobSubmittedEventHandler jseHandler = new JobSubmittedEventHandler("testId",
-        "testName", "testNodeName", "\"key2\"=\"value2\" \"key1\"=\"value1\" ");
+        "testName", "testNodeName", "\"key2\"=\"value2\" \"key1\"=\"value1\" ",
+        "tag1,tag2");
     dispatcher.register(EventType.class, jseHandler);
     JobImpl job = createStubbedJob(conf, dispatcher, 0);
     job.handle(new JobEvent(job.getID(), JobEventType.JOB_INIT));
@@ -706,14 +708,18 @@ public class TestJobImpl {
     
     private String workflowAdjacencies;
     
+    private String workflowTags;
+    
     private Boolean assertBoolean;
 
     public JobSubmittedEventHandler(String workflowId, String workflowName,
-        String workflowNodeName, String workflowAdjacencies) {
+        String workflowNodeName, String workflowAdjacencies,
+        String workflowTags) {
       this.workflowId = workflowId;
       this.workflowName = workflowName;
       this.workflowNodeName = workflowNodeName;
       this.workflowAdjacencies = workflowAdjacencies;
+      this.workflowTags = workflowTags;
       assertBoolean = null;
     }
 
@@ -739,6 +745,10 @@ public class TestJobImpl {
         setAssertValue(false);
         return;
       }
+      if (!workflowTags.equals(jsEvent.getWorkflowTags())) {
+        setAssertValue(false);
+        return;
+      }
       setAssertValue(true);
     }
     

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-common/src/main/java/org/apache/hadoop/mapred/LocalDistributedCacheManager.java Wed Apr 10 20:17:39 2013
@@ -32,13 +32,13 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.Random;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -91,6 +91,9 @@ class LocalDistributedCacheManager {
     Map<String, LocalResource> localResources = 
       new LinkedHashMap<String, LocalResource>();
     MRApps.setupDistributedCache(conf, localResources);
+    // Generating unique numbers for FSDownload.
+    AtomicLong uniqueNumberGenerator =
+        new AtomicLong(System.currentTimeMillis());
     
     // Find which resources are to be put on the local classpath
     Map<String, Path> classpaths = new HashMap<String, Path>();
@@ -128,8 +131,10 @@ class LocalDistributedCacheManager {
       Path destPath = localDirAllocator.getLocalPathForWrite(".", conf);
       Map<LocalResource, Future<Path>> resourcesToPaths = Maps.newHashMap();
       for (LocalResource resource : localResources.values()) {
-        Callable<Path> download = new FSDownload(localFSFileContext, ugi, conf,
-            destPath, resource, new Random());
+        Callable<Path> download =
+            new FSDownload(localFSFileContext, ugi, conf, new Path(destPath,
+                Long.toString(uniqueNumberGenerator.incrementAndGet())),
+                resource);
         Future<Path> future = exec.submit(download);
         resourcesToPaths.put(resource, future);
       }

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/avro/Events.avpr
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/avro/Events.avpr?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/avro/Events.avpr (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/avro/Events.avpr Wed Apr 10 20:17:39 2013
@@ -95,7 +95,8 @@
           {"name": "workflowId", "type": "string"},
           {"name": "workflowName", "type": "string"},
           {"name": "workflowNodeName", "type": "string"},
-          {"name": "workflowAdjacencies", "type": "string"}
+          {"name": "workflowAdjacencies", "type": "string"},
+          {"name": "workflowTags", "type": "string"}
       ]
      },
 

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java Wed Apr 10 20:17:39 2013
@@ -664,6 +664,8 @@ public interface MRJobConfig {
   public static final String WORKFLOW_ADJACENCY_PREFIX_PATTERN =
       "^mapreduce\\.workflow\\.adjacency\\..+";
 
+  public static final String WORKFLOW_TAGS = "mapreduce.workflow.tags";
+
   /**
    * The maximum number of application attempts.
    * It is a application-specific setting.

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobSubmittedEvent.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobSubmittedEvent.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobSubmittedEvent.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobSubmittedEvent.java Wed Apr 10 20:17:39 2013
@@ -75,6 +75,31 @@ public class JobSubmittedEvent implement
       Map<JobACL, AccessControlList> jobACLs, String jobQueueName,
       String workflowId, String workflowName, String workflowNodeName,
       String workflowAdjacencies) {
+    this(id, jobName, userName, submitTime, jobConfPath, jobACLs,
+        jobQueueName, workflowId, workflowName, workflowNodeName,
+        workflowAdjacencies, "");
+  }
+
+    /**
+     * Create an event to record job submission
+     * @param id The job Id of the job
+     * @param jobName Name of the job
+     * @param userName Name of the user who submitted the job
+     * @param submitTime Time of submission
+     * @param jobConfPath Path of the Job Configuration file
+     * @param jobACLs The configured acls for the job.
+     * @param jobQueueName The job-queue to which this job was submitted to
+     * @param workflowId The Id of the workflow
+     * @param workflowName The name of the workflow
+     * @param workflowNodeName The node name of the workflow
+     * @param workflowAdjacencies The adjacencies of the workflow
+     * @param workflowTags Comma-separated tags for the workflow
+     */
+    public JobSubmittedEvent(JobID id, String jobName, String userName,
+        long submitTime, String jobConfPath,
+        Map<JobACL, AccessControlList> jobACLs, String jobQueueName,
+        String workflowId, String workflowName, String workflowNodeName,
+        String workflowAdjacencies, String workflowTags) {
     datum.jobid = new Utf8(id.toString());
     datum.jobName = new Utf8(jobName);
     datum.userName = new Utf8(userName);
@@ -101,6 +126,9 @@ public class JobSubmittedEvent implement
     if (workflowAdjacencies != null) {
       datum.workflowAdjacencies = new Utf8(workflowAdjacencies);
     }
+    if (workflowTags != null) {
+      datum.workflowTags = new Utf8(workflowTags);
+    }
   }
 
   JobSubmittedEvent() {}
@@ -168,6 +196,13 @@ public class JobSubmittedEvent implement
     }
     return null;
   }
+  /** Get the workflow tags */
+  public String getWorkflowTags() {
+    if (datum.workflowTags != null) {
+      return datum.workflowTags.toString();
+    }
+    return null;
+  }
   /** Get the event type */
   public EventType getEventType() { return EventType.JOB_SUBMITTED; }
 

Propchange: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
------------------------------------------------------------------------------
  Merged /hadoop/common/trunk/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml:r1464808-1466652

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryFileManager.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryFileManager.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryFileManager.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/main/java/org/apache/hadoop/mapreduce/v2/hs/HistoryFileManager.java Wed Apr 10 20:17:39 2013
@@ -869,4 +869,9 @@ public class HistoryFileManager extends 
       }
     }
   }
+  // for test
+  @VisibleForTesting
+  void setMaxHistoryAge(long newValue){
+    maxHistoryAge=newValue;
+  } 
 }

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestCompletedTask.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestCompletedTask.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestCompletedTask.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestCompletedTask.java Wed Apr 10 20:17:39 2013
@@ -21,46 +21,75 @@ package org.apache.hadoop.mapreduce.v2.h
 import java.util.Map;
 import java.util.TreeMap;
 
+import org.apache.hadoop.mapreduce.JobID;
 import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskID;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskAttemptInfo;
 import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskInfo;
+import org.apache.hadoop.mapreduce.v2.api.records.Phase;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskId;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskReport;
 import org.apache.hadoop.mapreduce.v2.hs.CompletedTask;
-import org.junit.Assert;
 import org.junit.Test;
-import org.mockito.Mockito;
+import static org.mockito.Mockito.*;
+import static org.junit.Assert.*;
 
 public class TestCompletedTask{
 
-  @Test
+  @Test (timeout=5000)
   public void testTaskStartTimes() {
     
-    TaskId taskId = Mockito.mock(TaskId.class); 
-    TaskInfo taskInfo = Mockito.mock(TaskInfo.class);
+    TaskId taskId = mock(TaskId.class); 
+    TaskInfo taskInfo = mock(TaskInfo.class);
     Map<TaskAttemptID, TaskAttemptInfo> taskAttempts
       = new TreeMap<TaskAttemptID, TaskAttemptInfo>();
     
     TaskAttemptID id = new TaskAttemptID("0", 0, TaskType.MAP, 0, 0);
-    TaskAttemptInfo info = Mockito.mock(TaskAttemptInfo.class);
-    Mockito.when(info.getAttemptId()).thenReturn(id);
-    Mockito.when(info.getStartTime()).thenReturn(10l);
+    TaskAttemptInfo info = mock(TaskAttemptInfo.class);
+    when(info.getAttemptId()).thenReturn(id);
+    when(info.getStartTime()).thenReturn(10l);
     taskAttempts.put(id, info);
     
     id = new TaskAttemptID("1", 0, TaskType.MAP, 1, 1);
-    info = Mockito.mock(TaskAttemptInfo.class);
-    Mockito.when(info.getAttemptId()).thenReturn(id);
-    Mockito.when(info.getStartTime()).thenReturn(20l);
+    info = mock(TaskAttemptInfo.class);
+    when(info.getAttemptId()).thenReturn(id);
+    when(info.getStartTime()).thenReturn(20l);
     taskAttempts.put(id, info);
     
     
-    Mockito.when(taskInfo.getAllTaskAttempts()).thenReturn(taskAttempts);
+    when(taskInfo.getAllTaskAttempts()).thenReturn(taskAttempts);
     CompletedTask task = new CompletedTask(taskId, taskInfo);
     TaskReport report = task.getReport();
 
     // Make sure the startTime returned by report is the lesser of the 
     // attempy launch times
-    Assert.assertTrue(report.getStartTime() == 10);
+    assertTrue(report.getStartTime() == 10);
+  }
+  /**
+   * test some methods of CompletedTaskAttempt
+   */
+  @Test (timeout=5000)
+  public void testCompletedTaskAttempt(){
+    
+    TaskAttemptInfo attemptInfo= mock(TaskAttemptInfo.class);
+    when(attemptInfo.getRackname()).thenReturn("Rackname");
+    when(attemptInfo.getShuffleFinishTime()).thenReturn(11L);
+    when(attemptInfo.getSortFinishTime()).thenReturn(12L);
+    when(attemptInfo.getShufflePort()).thenReturn(10);
+    
+    JobID jobId= new JobID("12345",0);
+    TaskID taskId =new TaskID(jobId,TaskType.REDUCE, 0);
+    TaskAttemptID taskAttemptId= new TaskAttemptID(taskId, 0);
+    when(attemptInfo.getAttemptId()).thenReturn(taskAttemptId);
+    
+    
+    CompletedTaskAttempt taskAttemt= new CompletedTaskAttempt(null,attemptInfo);
+    assertEquals( "Rackname",   taskAttemt.getNodeRackName());
+    assertEquals( Phase.CLEANUP,   taskAttemt.getPhase());
+    assertTrue(  taskAttemt.isFinished());
+    assertEquals( 11L,   taskAttemt.getShuffleFinishTime());
+    assertEquals( 12L,   taskAttemt.getSortFinishTime());
+    assertEquals( 10,   taskAttemt.getShufflePort());
   }
 }

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryEntities.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryEntities.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryEntities.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryEntities.java Wed Apr 10 20:17:39 2013
@@ -45,7 +45,9 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
 
+import static org.junit.Assert.*;
 import static org.mockito.Mockito.*;
 
 @RunWith(value = Parameterized.class)
@@ -79,7 +81,7 @@ public class TestJobHistoryEntities {
   }
 
   /* Verify some expected values based on the history file */
-  @Test (timeout=10000)
+  @Test (timeout=100000)
   public void testCompletedJob() throws Exception {
     HistoryFileInfo info = mock(HistoryFileInfo.class);
     when(info.getConfFile()).thenReturn(fullConfPath);
@@ -168,4 +170,45 @@ public class TestJobHistoryEntities {
     assertEquals(45454, rta1Report.getNodeManagerPort());
     assertEquals(9999, rta1Report.getNodeManagerHttpPort());
   }
+  /**
+   * Simple test of some methods of CompletedJob
+   * @throws Exception
+   */
+  @Test (timeout=30000)
+  public void testGetTaskAttemptCompletionEvent() throws Exception{
+    HistoryFileInfo info = mock(HistoryFileInfo.class);
+    when(info.getConfFile()).thenReturn(fullConfPath);
+    completedJob =
+      new CompletedJob(conf, jobId, fulleHistoryPath, loadTasks, "user",
+          info, jobAclsManager);
+    TaskCompletionEvent[] events= completedJob.getMapAttemptCompletionEvents(0,1000);
+    assertEquals(10, completedJob.getMapAttemptCompletionEvents(0,10).length);
+    int currentEventId=0;
+    for (TaskCompletionEvent taskAttemptCompletionEvent : events) {
+      int eventId= taskAttemptCompletionEvent.getEventId();
+      assertTrue(eventId>=currentEventId);
+      currentEventId=eventId;
+    }
+    assertNull(completedJob.loadConfFile() );
+    // job name
+    assertEquals("Sleep job",completedJob.getName());
+    // queue name
+    assertEquals("default",completedJob.getQueueName());
+    // progress
+    assertEquals(1.0, completedJob.getProgress(),0.001);
+    // 11 rows in answer
+    assertEquals(11,completedJob.getTaskAttemptCompletionEvents(0,1000).length);
+    // select first 10 rows
+    assertEquals(10,completedJob.getTaskAttemptCompletionEvents(0,10).length);
+    // select 5-10 rows include 5th
+    assertEquals(6,completedJob.getTaskAttemptCompletionEvents(5,10).length);
+
+    // without errors
+    assertEquals(1,completedJob.getDiagnostics().size());
+    assertEquals("",completedJob.getDiagnostics().get(0));
+
+    assertEquals(0, completedJob.getJobACLs().size());
+
+  }
+
 }

Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryParsing.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryParsing.java?rev=1466658&r1=1466657&r2=1466658&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryParsing.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-hs/src/test/java/org/apache/hadoop/mapreduce/v2/hs/TestJobHistoryParsing.java Wed Apr 10 20:17:39 2013
@@ -19,6 +19,9 @@
 package org.apache.hadoop.mapreduce.v2.hs;
 
 import java.io.ByteArrayOutputStream;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
 import java.io.IOException;
 import java.io.PrintStream;
 import java.util.Arrays;
@@ -54,6 +57,9 @@ import org.apache.hadoop.mapreduce.v2.ap
 import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptId;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskId;
 import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
+import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
+import org.apache.hadoop.mapreduce.v2.api.records.impl.pb.JobIdPBImpl;
+import org.apache.hadoop.mapreduce.v2.api.records.impl.pb.TaskIdPBImpl;
 import org.apache.hadoop.mapreduce.v2.app.MRApp;
 import org.apache.hadoop.mapreduce.v2.app.job.Job;
 import org.apache.hadoop.mapreduce.v2.app.job.Task;
@@ -65,7 +71,9 @@ import org.apache.hadoop.mapreduce.v2.hs
 import org.apache.hadoop.mapreduce.v2.jobhistory.FileNameIndexUtils;
 import org.apache.hadoop.mapreduce.v2.jobhistory.JobHistoryUtils;
 import org.apache.hadoop.mapreduce.v2.jobhistory.JobIndexInfo;
+import org.apache.hadoop.mapreduce.v2.hs.webapp.dao.JobsInfo;
 import org.apache.hadoop.net.DNSToSwitchMapping;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.service.Service;
 import org.apache.hadoop.yarn.util.BuilderUtils;
@@ -80,12 +88,12 @@ public class TestJobHistoryParsing {
 
   private static final String RACK_NAME = "/MyRackName";
 
-  private  ByteArrayOutputStream outContent = new ByteArrayOutputStream();
+  private ByteArrayOutputStream outContent = new ByteArrayOutputStream();
 
   public static class MyResolver implements DNSToSwitchMapping {
     @Override
     public List<String> resolve(List<String> names) {
-      return Arrays.asList(new String[]{RACK_NAME});
+      return Arrays.asList(new String[] { RACK_NAME });
     }
 
     @Override
@@ -93,14 +101,14 @@ public class TestJobHistoryParsing {
     }
   }
 
-  @Test (timeout=50000)
+  @Test(timeout = 50000)
   public void testJobInfo() throws Exception {
     JobInfo info = new JobInfo();
     Assert.assertEquals("NORMAL", info.getPriority());
     info.printAll();
   }
 
-  @Test (timeout=50000)
+  @Test(timeout = 300000)
   public void testHistoryParsing() throws Exception {
     LOG.info("STARTING testHistoryParsing()");
     try {
@@ -109,8 +117,8 @@ public class TestJobHistoryParsing {
       LOG.info("FINISHED testHistoryParsing()");
     }
   }
-  
-  @Test (timeout=50000)
+
+  @Test(timeout = 50000)
   public void testHistoryParsingWithParseErrors() throws Exception {
     LOG.info("STARTING testHistoryParsingWithParseErrors()");
     try {
@@ -119,18 +127,18 @@ public class TestJobHistoryParsing {
       LOG.info("FINISHED testHistoryParsingWithParseErrors()");
     }
   }
-  
-  private static String getJobSummary(FileContext fc, Path path) throws IOException {
+
+  private static String getJobSummary(FileContext fc, Path path)
+      throws IOException {
     Path qPath = fc.makeQualified(path);
     FSDataInputStream in = fc.open(qPath);
     String jobSummaryString = in.readUTF();
     in.close();
     return jobSummaryString;
   }
-  
+
   private void checkHistoryParsing(final int numMaps, final int numReduces,
-      final int numSuccessfulMaps) 
-  throws Exception {
+      final int numSuccessfulMaps) throws Exception {
     Configuration conf = new Configuration();
     conf.set(MRJobConfig.USER_NAME, System.getProperty("user.name"));
     long amStartTimeEst = System.currentTimeMillis();
@@ -138,9 +146,8 @@ public class TestJobHistoryParsing {
         CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
         MyResolver.class, DNSToSwitchMapping.class);
     RackResolver.init(conf);
-    MRApp app = 
-        new MRAppWithHistory(numMaps, numReduces, true, 
-            this.getClass().getName(), true);
+    MRApp app = new MRAppWithHistory(numMaps, numReduces, true, this.getClass()
+        .getName(), true);
     app.submit(conf);
     Job job = app.getContext().getAllJobs().values().iterator().next();
     JobId jobId = job.getID();
@@ -152,7 +159,7 @@ public class TestJobHistoryParsing {
 
     String jobhistoryDir = JobHistoryUtils
         .getHistoryIntermediateDoneDirForUser(conf);
-    
+
     FileContext fc = null;
     try {
       fc = FileContext.getFileContext(conf);
@@ -160,7 +167,7 @@ public class TestJobHistoryParsing {
       LOG.info("Can not get FileContext", ioe);
       throw (new Exception("Can not get File Context"));
     }
-    
+
     if (numMaps == numSuccessfulMaps) {
       String summaryFileName = JobHistoryUtils
           .getIntermediateSummaryFileName(jobId);
@@ -185,20 +192,22 @@ public class TestJobHistoryParsing {
           Long.parseLong(jobSummaryElements.get("submitTime")) != 0);
       Assert.assertTrue("launchTime should not be 0",
           Long.parseLong(jobSummaryElements.get("launchTime")) != 0);
-      Assert.assertTrue("firstMapTaskLaunchTime should not be 0",
-          Long.parseLong(jobSummaryElements.get("firstMapTaskLaunchTime")) != 0);
       Assert
-      .assertTrue(
-          "firstReduceTaskLaunchTime should not be 0",
-          Long.parseLong(jobSummaryElements.get("firstReduceTaskLaunchTime")) != 0);
+          .assertTrue(
+              "firstMapTaskLaunchTime should not be 0",
+              Long.parseLong(jobSummaryElements.get("firstMapTaskLaunchTime")) != 0);
+      Assert
+          .assertTrue("firstReduceTaskLaunchTime should not be 0",
+              Long.parseLong(jobSummaryElements
+                  .get("firstReduceTaskLaunchTime")) != 0);
       Assert.assertTrue("finishTime should not be 0",
           Long.parseLong(jobSummaryElements.get("finishTime")) != 0);
       Assert.assertEquals("Mismatch in num map slots", numSuccessfulMaps,
           Integer.parseInt(jobSummaryElements.get("numMaps")));
       Assert.assertEquals("Mismatch in num reduce slots", numReduces,
           Integer.parseInt(jobSummaryElements.get("numReduces")));
-      Assert.assertEquals("User does not match", System.getProperty("user.name"),
-          jobSummaryElements.get("user"));
+      Assert.assertEquals("User does not match",
+          System.getProperty("user.name"), jobSummaryElements.get("user"));
       Assert.assertEquals("Queue does not match", "default",
           jobSummaryElements.get("queue"));
       Assert.assertEquals("Status does not match", "SUCCEEDED",
@@ -210,8 +219,8 @@ public class TestJobHistoryParsing {
     HistoryFileInfo fileInfo = jobHistory.getJobFileInfo(jobId);
     JobInfo jobInfo;
     long numFinishedMaps;
-    
-    synchronized(fileInfo) {
+
+    synchronized (fileInfo) {
       Path historyFilePath = fileInfo.getHistoryFile();
       FSDataInputStream in = null;
       LOG.info("JobHistoryFile is: " + historyFilePath);
@@ -228,11 +237,11 @@ public class TestJobHistoryParsing {
       if (numMaps == numSuccessfulMaps) {
         reader = realReader;
       } else {
-        final AtomicInteger numFinishedEvents = new AtomicInteger(0);  // Hack!
+        final AtomicInteger numFinishedEvents = new AtomicInteger(0); // Hack!
         Mockito.when(reader.getNextEvent()).thenAnswer(
             new Answer<HistoryEvent>() {
-              public HistoryEvent answer(InvocationOnMock invocation) 
-              throws IOException {
+              public HistoryEvent answer(InvocationOnMock invocation)
+                  throws IOException {
                 HistoryEvent event = realReader.getNextEvent();
                 if (event instanceof TaskFinishedEvent) {
                   numFinishedEvents.incrementAndGet();
@@ -244,22 +253,20 @@ public class TestJobHistoryParsing {
                   throw new IOException("test");
                 }
               }
-            }
-        );
+            });
       }
 
       jobInfo = parser.parse(reader);
 
-      numFinishedMaps = 
-        computeFinishedMaps(jobInfo, numMaps, numSuccessfulMaps);
+      numFinishedMaps = computeFinishedMaps(jobInfo, numMaps, numSuccessfulMaps);
 
       if (numFinishedMaps != numMaps) {
         Exception parseException = parser.getParseException();
-        Assert.assertNotNull("Didn't get expected parse exception", 
+        Assert.assertNotNull("Didn't get expected parse exception",
             parseException);
       }
     }
-    
+
     Assert.assertEquals("Incorrect username ", System.getProperty("user.name"),
         jobInfo.getUsername());
     Assert.assertEquals("Incorrect jobName ", "test", jobInfo.getJobname());
@@ -267,7 +274,7 @@ public class TestJobHistoryParsing {
         jobInfo.getJobQueueName());
     Assert
         .assertEquals("incorrect conf path", "test", jobInfo.getJobConfPath());
-    Assert.assertEquals("incorrect finishedMap ", numSuccessfulMaps, 
+    Assert.assertEquals("incorrect finishedMap ", numSuccessfulMaps,
         numFinishedMaps);
     Assert.assertEquals("incorrect finishedReduces ", numReduces,
         jobInfo.getFinishedReduces());
@@ -275,8 +282,8 @@ public class TestJobHistoryParsing {
         jobInfo.getUberized());
     Map<TaskID, TaskInfo> allTasks = jobInfo.getAllTasks();
     int totalTasks = allTasks.size();
-    Assert.assertEquals("total number of tasks is incorrect  ", 
-        (numMaps+numReduces), totalTasks);
+    Assert.assertEquals("total number of tasks is incorrect  ",
+        (numMaps + numReduces), totalTasks);
 
     // Verify aminfo
     Assert.assertEquals(1, jobInfo.getAMInfos().size());
@@ -306,8 +313,7 @@ public class TestJobHistoryParsing {
 
     // Deep compare Job and JobInfo
     for (Task task : job.getTasks().values()) {
-      TaskInfo taskInfo = allTasks.get(
-          TypeConverter.fromYarn(task.getID()));
+      TaskInfo taskInfo = allTasks.get(TypeConverter.fromYarn(task.getID()));
       Assert.assertNotNull("TaskInfo not found", taskInfo);
       for (TaskAttempt taskAttempt : task.getAttempts().values()) {
         TaskAttemptInfo taskAttemptInfo = taskInfo.getAllTaskAttempts().get(
@@ -318,27 +324,32 @@ public class TestJobHistoryParsing {
         if (numMaps == numSuccessfulMaps) {
           Assert.assertEquals(MRApp.NM_HOST, taskAttemptInfo.getHostname());
           Assert.assertEquals(MRApp.NM_PORT, taskAttemptInfo.getPort());
-          
+
           // Verify rack-name
-          Assert.assertEquals("rack-name is incorrect", taskAttemptInfo
-              .getRackname(), RACK_NAME);
+          Assert.assertEquals("rack-name is incorrect",
+              taskAttemptInfo.getRackname(), RACK_NAME);
         }
       }
     }
-    
+
     // test output for HistoryViewer
-    PrintStream stdps=System.out;
+    PrintStream stdps = System.out;
     try {
       System.setOut(new PrintStream(outContent));
       HistoryViewer viewer = new HistoryViewer(fc.makeQualified(
           fileInfo.getHistoryFile()).toString(), conf, true);
       viewer.print();
-      
-      for (TaskInfo taskInfo : allTasks.values()) { 
-        
-        String test=  (taskInfo.getTaskStatus()==null?"":taskInfo.getTaskStatus())+" "+taskInfo.getTaskType()+" task list for "+taskInfo.getTaskId().getJobID();
-        Assert.assertTrue(outContent.toString().indexOf(test)>0);
-        Assert.assertTrue(outContent.toString().indexOf(taskInfo.getTaskId().toString())>0);
+
+      for (TaskInfo taskInfo : allTasks.values()) {
+
+        String test = (taskInfo.getTaskStatus() == null ? "" : taskInfo
+            .getTaskStatus())
+            + " "
+            + taskInfo.getTaskType()
+            + " task list for " + taskInfo.getTaskId().getJobID();
+        Assert.assertTrue(outContent.toString().indexOf(test) > 0);
+        Assert.assertTrue(outContent.toString().indexOf(
+            taskInfo.getTaskId().toString()) > 0);
       }
     } finally {
       System.setOut(stdps);
@@ -363,186 +374,180 @@ public class TestJobHistoryParsing {
     }
     return numFinishedMaps;
   }
-  
-  @Test (timeout=50000)
+
+  @Test(timeout = 30000)
   public void testHistoryParsingForFailedAttempts() throws Exception {
     LOG.info("STARTING testHistoryParsingForFailedAttempts");
     try {
-    Configuration conf = new Configuration();
-    conf
-        .setClass(
-            CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
-            MyResolver.class, DNSToSwitchMapping.class);
-    RackResolver.init(conf);
-    MRApp app = new MRAppWithHistoryWithFailedAttempt(2, 1, true, this.getClass().getName(),
-        true);
-    app.submit(conf);
-    Job job = app.getContext().getAllJobs().values().iterator().next();
-    JobId jobId = job.getID();
-    app.waitForState(job, JobState.SUCCEEDED);
-    
-    // make sure all events are flushed
-    app.waitForState(Service.STATE.STOPPED);
-
-    String jobhistoryDir = JobHistoryUtils
-        .getHistoryIntermediateDoneDirForUser(conf);
-    JobHistory jobHistory = new JobHistory();
-    jobHistory.init(conf);
+      Configuration conf = new Configuration();
+      conf.setClass(
+          CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+          MyResolver.class, DNSToSwitchMapping.class);
+      RackResolver.init(conf);
+      MRApp app = new MRAppWithHistoryWithFailedAttempt(2, 1, true, this
+          .getClass().getName(), true);
+      app.submit(conf);
+      Job job = app.getContext().getAllJobs().values().iterator().next();
+      JobId jobId = job.getID();
+      app.waitForState(job, JobState.SUCCEEDED);
+
+      // make sure all events are flushed
+      app.waitForState(Service.STATE.STOPPED);
+
+      String jobhistoryDir = JobHistoryUtils
+          .getHistoryIntermediateDoneDirForUser(conf);
+      JobHistory jobHistory = new JobHistory();
+      jobHistory.init(conf);
+
+      JobIndexInfo jobIndexInfo = jobHistory.getJobFileInfo(jobId)
+          .getJobIndexInfo();
+      String jobhistoryFileName = FileNameIndexUtils
+          .getDoneFileName(jobIndexInfo);
 
-    JobIndexInfo jobIndexInfo = jobHistory.getJobFileInfo(jobId)
-        .getJobIndexInfo();
-    String jobhistoryFileName = FileNameIndexUtils
-        .getDoneFileName(jobIndexInfo);
-
-    Path historyFilePath = new Path(jobhistoryDir, jobhistoryFileName);
-    FSDataInputStream in = null;
-    FileContext fc = null;
-    try {
-      fc = FileContext.getFileContext(conf);
-      in = fc.open(fc.makeQualified(historyFilePath));
-    } catch (IOException ioe) {
-      LOG.info("Can not open history file: " + historyFilePath, ioe);
-      throw (new Exception("Can not open History File"));
-    }
+      Path historyFilePath = new Path(jobhistoryDir, jobhistoryFileName);
+      FSDataInputStream in = null;
+      FileContext fc = null;
+      try {
+        fc = FileContext.getFileContext(conf);
+        in = fc.open(fc.makeQualified(historyFilePath));
+      } catch (IOException ioe) {
+        LOG.info("Can not open history file: " + historyFilePath, ioe);
+        throw (new Exception("Can not open History File"));
+      }
 
-    JobHistoryParser parser = new JobHistoryParser(in);
-    JobInfo jobInfo = parser.parse();
-    Exception parseException = parser.getParseException();
-    Assert.assertNull("Caught an expected exception " + parseException, 
-        parseException);
-    int noOffailedAttempts = 0;
-    Map<TaskID, TaskInfo> allTasks = jobInfo.getAllTasks();
-    for (Task task : job.getTasks().values()) {
-      TaskInfo taskInfo = allTasks.get(TypeConverter.fromYarn(task.getID()));
-      for (TaskAttempt taskAttempt : task.getAttempts().values()) {
-        TaskAttemptInfo taskAttemptInfo = taskInfo.getAllTaskAttempts().get(
-            TypeConverter.fromYarn((taskAttempt.getID())));
-        // Verify rack-name for all task attempts
-        Assert.assertEquals("rack-name is incorrect", taskAttemptInfo
-            .getRackname(), RACK_NAME);
-        if (taskAttemptInfo.getTaskStatus().equals("FAILED")) {
-          noOffailedAttempts++;
+      JobHistoryParser parser = new JobHistoryParser(in);
+      JobInfo jobInfo = parser.parse();
+      Exception parseException = parser.getParseException();
+      Assert.assertNull("Caught an expected exception " + parseException,
+          parseException);
+      int noOffailedAttempts = 0;
+      Map<TaskID, TaskInfo> allTasks = jobInfo.getAllTasks();
+      for (Task task : job.getTasks().values()) {
+        TaskInfo taskInfo = allTasks.get(TypeConverter.fromYarn(task.getID()));
+        for (TaskAttempt taskAttempt : task.getAttempts().values()) {
+          TaskAttemptInfo taskAttemptInfo = taskInfo.getAllTaskAttempts().get(
+              TypeConverter.fromYarn((taskAttempt.getID())));
+          // Verify rack-name for all task attempts
+          Assert.assertEquals("rack-name is incorrect",
+              taskAttemptInfo.getRackname(), RACK_NAME);
+          if (taskAttemptInfo.getTaskStatus().equals("FAILED")) {
+            noOffailedAttempts++;
+          }
         }
       }
-    }
-    Assert.assertEquals("No of Failed tasks doesn't match.", 2, noOffailedAttempts);
+      Assert.assertEquals("No of Failed tasks doesn't match.", 2,
+          noOffailedAttempts);
     } finally {
       LOG.info("FINISHED testHistoryParsingForFailedAttempts");
     }
   }
-  
-  @Test (timeout=5000)
+
+  @Test(timeout = 60000)
   public void testCountersForFailedTask() throws Exception {
     LOG.info("STARTING testCountersForFailedTask");
     try {
-    Configuration conf = new Configuration();
-    conf
-        .setClass(
-            CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
-            MyResolver.class, DNSToSwitchMapping.class);
-    RackResolver.init(conf);
-    MRApp app = new MRAppWithHistoryWithFailedTask(2, 1, true,
-        this.getClass().getName(), true);
-    app.submit(conf);
-    Job job = app.getContext().getAllJobs().values().iterator().next();
-    JobId jobId = job.getID();
-    app.waitForState(job, JobState.FAILED);
-
-    // make sure all events are flushed
-    app.waitForState(Service.STATE.STOPPED);
+      Configuration conf = new Configuration();
+      conf.setClass(
+          CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+          MyResolver.class, DNSToSwitchMapping.class);
+      RackResolver.init(conf);
+      MRApp app = new MRAppWithHistoryWithFailedTask(2, 1, true, this
+          .getClass().getName(), true);
+      app.submit(conf);
+      Job job = app.getContext().getAllJobs().values().iterator().next();
+      JobId jobId = job.getID();
+      app.waitForState(job, JobState.FAILED);
+
+      // make sure all events are flushed
+      app.waitForState(Service.STATE.STOPPED);
+
+      String jobhistoryDir = JobHistoryUtils
+          .getHistoryIntermediateDoneDirForUser(conf);
+      JobHistory jobHistory = new JobHistory();
+      jobHistory.init(conf);
+
+      JobIndexInfo jobIndexInfo = jobHistory.getJobFileInfo(jobId)
+          .getJobIndexInfo();
+      String jobhistoryFileName = FileNameIndexUtils
+          .getDoneFileName(jobIndexInfo);
 
-    String jobhistoryDir = JobHistoryUtils
-        .getHistoryIntermediateDoneDirForUser(conf);
-    JobHistory jobHistory = new JobHistory();
-    jobHistory.init(conf);
-
-    JobIndexInfo jobIndexInfo = jobHistory.getJobFileInfo(jobId)
-        .getJobIndexInfo();
-    String jobhistoryFileName = FileNameIndexUtils
-        .getDoneFileName(jobIndexInfo);
-
-    Path historyFilePath = new Path(jobhistoryDir, jobhistoryFileName);
-    FSDataInputStream in = null;
-    FileContext fc = null;
-    try {
-      fc = FileContext.getFileContext(conf);
-      in = fc.open(fc.makeQualified(historyFilePath));
-    } catch (IOException ioe) {
-      LOG.info("Can not open history file: " + historyFilePath, ioe);
-      throw (new Exception("Can not open History File"));
-    }
+      Path historyFilePath = new Path(jobhistoryDir, jobhistoryFileName);
+      FSDataInputStream in = null;
+      FileContext fc = null;
+      try {
+        fc = FileContext.getFileContext(conf);
+        in = fc.open(fc.makeQualified(historyFilePath));
+      } catch (IOException ioe) {
+        LOG.info("Can not open history file: " + historyFilePath, ioe);
+        throw (new Exception("Can not open History File"));
+      }
 
-    JobHistoryParser parser = new JobHistoryParser(in);
-    JobInfo jobInfo = parser.parse();
-    Exception parseException = parser.getParseException();
-    Assert.assertNull("Caught an expected exception " + parseException,
-        parseException);
-    for (Map.Entry<TaskID,TaskInfo> entry : jobInfo.getAllTasks().entrySet()) {
-      TaskId yarnTaskID = TypeConverter.toYarn(entry.getKey());
-      CompletedTask ct = new CompletedTask(yarnTaskID, entry.getValue());
-      Assert.assertNotNull("completed task report has null counters",
-          ct.getReport().getCounters());
-      //Make sure all the completedTask has counters, and the counters are not empty
-      Assert.assertTrue(ct.getReport().getCounters()
-          .getAllCounterGroups().size() > 0);
-    }
+      JobHistoryParser parser = new JobHistoryParser(in);
+      JobInfo jobInfo = parser.parse();
+      Exception parseException = parser.getParseException();
+      Assert.assertNull("Caught an expected exception " + parseException,
+          parseException);
+      for (Map.Entry<TaskID, TaskInfo> entry : jobInfo.getAllTasks().entrySet()) {
+        TaskId yarnTaskID = TypeConverter.toYarn(entry.getKey());
+        CompletedTask ct = new CompletedTask(yarnTaskID, entry.getValue());
+        Assert.assertNotNull("completed task report has null counters", ct
+            .getReport().getCounters());
+      }
     } finally {
       LOG.info("FINISHED testCountersForFailedTask");
     }
   }
 
-  @Test (timeout=50000)
+  @Test(timeout = 50000)
   public void testScanningOldDirs() throws Exception {
     LOG.info("STARTING testScanningOldDirs");
     try {
-    Configuration conf = new Configuration();
-    conf
-        .setClass(
-            CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
-            MyResolver.class, DNSToSwitchMapping.class);
-    RackResolver.init(conf);
-    MRApp app =
-        new MRAppWithHistory(1, 1, true,
-            this.getClass().getName(), true);
-    app.submit(conf);
-    Job job = app.getContext().getAllJobs().values().iterator().next();
-    JobId jobId = job.getID();
-    LOG.info("JOBID is " + TypeConverter.fromYarn(jobId).toString());
-    app.waitForState(job, JobState.SUCCEEDED);
-
-    // make sure all events are flushed
-    app.waitForState(Service.STATE.STOPPED);
+      Configuration conf = new Configuration();
+      conf.setClass(
+          CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+          MyResolver.class, DNSToSwitchMapping.class);
+      RackResolver.init(conf);
+      MRApp app = new MRAppWithHistory(1, 1, true, this.getClass().getName(),
+          true);
+      app.submit(conf);
+      Job job = app.getContext().getAllJobs().values().iterator().next();
+      JobId jobId = job.getID();
+      LOG.info("JOBID is " + TypeConverter.fromYarn(jobId).toString());
+      app.waitForState(job, JobState.SUCCEEDED);
+
+      // make sure all events are flushed
+      app.waitForState(Service.STATE.STOPPED);
+
+      HistoryFileManagerForTest hfm = new HistoryFileManagerForTest();
+      hfm.init(conf);
+      HistoryFileInfo fileInfo = hfm.getFileInfo(jobId);
+      Assert.assertNotNull("Unable to locate job history", fileInfo);
+
+      // force the manager to "forget" the job
+      hfm.deleteJobFromJobListCache(fileInfo);
+      final int msecPerSleep = 10;
+      int msecToSleep = 10 * 1000;
+      while (fileInfo.isMovePending() && msecToSleep > 0) {
+        Assert.assertTrue(!fileInfo.didMoveFail());
+        msecToSleep -= msecPerSleep;
+        Thread.sleep(msecPerSleep);
+      }
+      Assert.assertTrue("Timeout waiting for history move", msecToSleep > 0);
 
-    HistoryFileManagerForTest hfm = new HistoryFileManagerForTest();
-    hfm.init(conf);
-    HistoryFileInfo fileInfo = hfm.getFileInfo(jobId);
-    Assert.assertNotNull("Unable to locate job history", fileInfo);
-
-    // force the manager to "forget" the job
-    hfm.deleteJobFromJobListCache(fileInfo);
-    final int msecPerSleep = 10;
-    int msecToSleep = 10 * 1000;
-    while (fileInfo.isMovePending() && msecToSleep > 0) {
-      Assert.assertTrue(!fileInfo.didMoveFail());
-      msecToSleep -= msecPerSleep;
-      Thread.sleep(msecPerSleep);
-    }
-    Assert.assertTrue("Timeout waiting for history move", msecToSleep > 0);
-
-    fileInfo = hfm.getFileInfo(jobId);
-    Assert.assertNotNull("Unable to locate old job history", fileInfo);
-   } finally {
+      fileInfo = hfm.getFileInfo(jobId);
+      Assert.assertNotNull("Unable to locate old job history", fileInfo);
+    } finally {
       LOG.info("FINISHED testScanningOldDirs");
     }
   }
 
   static class MRAppWithHistoryWithFailedAttempt extends MRAppWithHistory {
 
-    public MRAppWithHistoryWithFailedAttempt(int maps, int reduces, boolean autoComplete,
-        String testName, boolean cleanOnStart) {
+    public MRAppWithHistoryWithFailedAttempt(int maps, int reduces,
+        boolean autoComplete, String testName, boolean cleanOnStart) {
       super(maps, reduces, autoComplete, testName, cleanOnStart);
     }
-    
+
     @SuppressWarnings("unchecked")
     @Override
     protected void attemptLaunched(TaskAttemptId attemptID) {
@@ -558,8 +563,8 @@ public class TestJobHistoryParsing {
 
   static class MRAppWithHistoryWithFailedTask extends MRAppWithHistory {
 
-    public MRAppWithHistoryWithFailedTask(int maps, int reduces, boolean autoComplete,
-        String testName, boolean cleanOnStart) {
+    public MRAppWithHistoryWithFailedTask(int maps, int reduces,
+        boolean autoComplete, String testName, boolean cleanOnStart) {
       super(maps, reduces, autoComplete, testName, cleanOnStart);
     }
 
@@ -587,4 +592,133 @@ public class TestJobHistoryParsing {
     t.testHistoryParsing();
     t.testHistoryParsingForFailedAttempts();
   }
+
+  /**
+   * Test clean old history files. Files should be deleted after 1 week by
+   * default.
+   */
+  @Test(timeout = 15000)
+  public void testDeleteFileInfo() throws Exception {
+    LOG.info("STARTING testDeleteFileInfo");
+    try {
+      Configuration conf = new Configuration();
+
+      conf.setClass(
+          CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+          MyResolver.class, DNSToSwitchMapping.class);
+
+      RackResolver.init(conf);
+      MRApp app = new MRAppWithHistory(1, 1, true, this.getClass().getName(),
+          true);
+      app.submit(conf);
+      Job job = app.getContext().getAllJobs().values().iterator().next();
+      JobId jobId = job.getID();
+
+      app.waitForState(job, JobState.SUCCEEDED);
+
+      // make sure all events are flushed
+      app.waitForState(Service.STATE.STOPPED);
+
+      HistoryFileManager hfm = new HistoryFileManager();
+      hfm.init(conf);
+      HistoryFileInfo fileInfo = hfm.getFileInfo(jobId);
+      hfm.initExisting();
+      // wait for move files form the done_intermediate directory to the gone
+      // directory
+      while (fileInfo.isMovePending()) {
+        Thread.sleep(300);
+      }
+
+      Assert.assertNotNull(hfm.jobListCache.values());
+
+      // try to remove fileInfo
+      hfm.clean();
+      // check that fileInfo does not deleted
+      Assert.assertFalse(fileInfo.isDeleted());
+      // correct live time
+      hfm.setMaxHistoryAge(-1);
+      hfm.clean();
+      // should be deleted !
+      Assert.assertTrue("file should be deleted ", fileInfo.isDeleted());
+
+    } finally {
+      LOG.info("FINISHED testDeleteFileInfo");
+    }
+  }
+
+  /**
+   * Simple test some methods of JobHistory
+   */
+  @Test(timeout = 20000)
+  public void testJobHistoryMethods() throws Exception {
+    LOG.info("STARTING testJobHistoryMethods");
+    try {
+      Configuration configuration = new Configuration();
+      configuration
+          .setClass(
+              CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+              MyResolver.class, DNSToSwitchMapping.class);
+
+      RackResolver.init(configuration);
+      MRApp app = new MRAppWithHistory(1, 1, true, this.getClass().getName(),
+          true);
+      app.submit(configuration);
+      Job job = app.getContext().getAllJobs().values().iterator().next();
+      JobId jobId = job.getID();
+      LOG.info("JOBID is " + TypeConverter.fromYarn(jobId).toString());
+      app.waitForState(job, JobState.SUCCEEDED);
+
+      JobHistory jobHistory = new JobHistory();
+      jobHistory.init(configuration);
+      // Method getAllJobs
+      Assert.assertEquals(1, jobHistory.getAllJobs().size());
+      // and with ApplicationId
+      Assert.assertEquals(1, jobHistory.getAllJobs(app.getAppID()).size());
+
+      JobsInfo jobsinfo = jobHistory.getPartialJobs(0L, 10L, null, "default",
+          0L, System.currentTimeMillis() + 1, 0L,
+          System.currentTimeMillis() + 1, JobState.SUCCEEDED);
+
+      Assert.assertEquals(1, jobsinfo.getJobs().size());
+      Assert.assertNotNull(jobHistory.getApplicationAttemptId());
+      // test Application Id
+      Assert.assertEquals("application_0_0000", jobHistory.getApplicationID()
+          .toString());
+      Assert
+          .assertEquals("Job History Server", jobHistory.getApplicationName());
+      // method does not work
+      Assert.assertNull(jobHistory.getEventHandler());
+      // method does not work
+      Assert.assertNull(jobHistory.getClock());
+      // method does not work
+      Assert.assertNull(jobHistory.getClusterInfo());
+
+    } finally {
+      LOG.info("FINISHED testJobHistoryMethods");
+    }
+  }
+
+  /**
+   * Simple test PartialJob
+   */
+  @Test(timeout = 1000)
+  public void testPartialJob() throws Exception {
+    JobId jobId = new JobIdPBImpl();
+    jobId.setId(0);
+    JobIndexInfo jii = new JobIndexInfo(0L, System.currentTimeMillis(), "user",
+        "jobName", jobId, 3, 2, "JobStatus");
+    PartialJob test = new PartialJob(jii, jobId);
+    
+    Assert.assertEquals(1.0f, test.getProgress(), 0.001f);
+    assertNull(test.getAllCounters());
+    assertNull(test.getTasks());
+    assertNull(test.getTasks(TaskType.MAP));
+    assertNull(test.getTask(new TaskIdPBImpl()));
+
+    assertNull(test.getTaskAttemptCompletionEvents(0, 100));
+    assertNull(test.getMapAttemptCompletionEvents(0, 100));
+    assertTrue(test.checkAccess(UserGroupInformation.getCurrentUser(), null));
+    assertNull(test.getAMInfos());
+
+  }
 }



Mime
View raw message