commit 3d4b39bfadfbc76f42aea7aef0cd3a402527078f

Author: Sharad Agarwal <sharadag@yahoo-inc.com>
Date:   Wed Mar 3 13:10:01 2010 +0530

    implement task verification utils


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

src/test/system/aop/org/apache/hadoop/mapred/TaskTrackerAspect.aj

@@ -73,14 +73,13 @@ public privileged aspect TaskTrackerAspect {
   private TTTaskInfo TaskTracker.getTTTaskInfo(TaskInProgress tip) {
     TTTaskInfo info;
     if (tip.task.isMapTask()) {
-      info =
-          new MapTTTaskInfo(tip.slotTaken, tip.wasKilled, tip.diagnosticInfo
-              .toString(), (MapTaskStatus) tip.getStatus());
+      info = new MapTTTaskInfo(tip.slotTaken, tip.wasKilled,
+          (MapTaskStatus) tip.getStatus(), tip.getJobConf(), tip.getTask()
+              .getUser(), tip.getTask().isTaskCleanupTask());
     } else {
-      info =
-          new ReduceTTTaskInfo(
-              tip.slotTaken, tip.wasKilled, tip.diagnosticInfo.toString(),
-              (ReduceTaskStatus) tip.getStatus());
+      info = new ReduceTTTaskInfo(tip.slotTaken, tip.wasKilled,
+          (ReduceTaskStatus) tip.getStatus(), tip.getJobConf(), tip.getTask()
+              .getUser(), tip.getTask().isTaskCleanupTask());
     }
     return info;
   }

src/test/system/java/org/apache/hadoop/mapred/TTTaskInfoImpl.java

@@ -22,6 +22,7 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapreduce.test.system.TTTaskInfo;
 /**
  * Abstract class which passes the Task view of the TaskTracker to the client.
@@ -30,26 +31,26 @@ import org.apache.hadoop.mapreduce.test.system.TTTaskInfo;
  */
 abstract class TTTaskInfoImpl implements TTTaskInfo {
 
-  private String diagonsticInfo;
   private boolean slotTaken;
   private boolean wasKilled;
   TaskStatus status;
+  Configuration conf;
+  String user;
+  boolean isTaskCleanupTask;
 
   public TTTaskInfoImpl() {
   }
 
   public TTTaskInfoImpl(boolean slotTaken, boolean wasKilled,
-      String diagonsticInfo, TaskStatus status) {
+      TaskStatus status, Configuration conf, String user,
+      boolean isTaskCleanupTask) {
     super();
-    this.diagonsticInfo = diagonsticInfo;
     this.slotTaken = slotTaken;
     this.wasKilled = wasKilled;
     this.status = status;
-  }
-
-  @Override
-  public String getDiagnosticInfo() {
-    return diagonsticInfo;
+    this.conf = conf;
+    this.user = user;
+    this.isTaskCleanupTask = isTaskCleanupTask;
   }
 
   @Override
@@ -63,47 +64,88 @@ abstract class TTTaskInfoImpl implements TTTaskInfo {
   }
 
   @Override
-  public TaskStatus getTaskStatus() {
-    return status;
-  }
+  public abstract TaskStatus getTaskStatus();
 
+  @Override
+  public Configuration getConf() {
+    return conf;
+  }
+  
+  @Override
+  public String getUser() {
+    return user;
+  }
+  
+  @Override
+  public boolean isTaskCleanupTask() {
+    return isTaskCleanupTask;
+  }
+  
   @Override
   public void readFields(DataInput in) throws IOException {
     slotTaken = in.readBoolean();
     wasKilled = in.readBoolean();
-    diagonsticInfo = in.readUTF();
-    status.readFields(in);
+    conf = new Configuration();
+    conf.readFields(in);
+    user = in.readUTF();
+    isTaskCleanupTask = in.readBoolean();
   }
 
   @Override
   public void write(DataOutput out) throws IOException {
     out.writeBoolean(slotTaken);
     out.writeBoolean(wasKilled);
-    out.writeUTF(diagonsticInfo);
+    conf.write(out);
+    out.writeUTF(user);
+    out.writeBoolean(isTaskCleanupTask);
     status.write(out);
   }
 
   static class MapTTTaskInfo extends TTTaskInfoImpl {
 
     public MapTTTaskInfo() {
-      super(false, false, "", new MapTaskStatus());
+      super();
     }
 
     public MapTTTaskInfo(boolean slotTaken, boolean wasKilled,
-        String diagonsticInfo, MapTaskStatus status) {
-      super(slotTaken, wasKilled, diagonsticInfo, status);
+        MapTaskStatus status, Configuration conf, String user,
+        boolean isTaskCleanup) {
+      super(slotTaken, wasKilled, status, conf, user, isTaskCleanup);
+    }
+
+    @Override
+    public TaskStatus getTaskStatus() {
+      return status;
+    }
+    
+    public void readFields(DataInput in) throws IOException {
+      super.readFields(in);
+      status = new MapTaskStatus();
+      status.readFields(in);
     }
   }
 
   static class ReduceTTTaskInfo extends TTTaskInfoImpl {
 
     public ReduceTTTaskInfo() {
-      super(false, false, "", new ReduceTaskStatus());
+      super();
+    }
+
+    public ReduceTTTaskInfo(boolean slotTaken, boolean wasKilled,
+        ReduceTaskStatus status, Configuration conf, String user,
+        boolean isTaskCleanup) {
+      super(slotTaken, wasKilled, status, conf, user, isTaskCleanup);
     }
 
-    public ReduceTTTaskInfo(boolean slotTaken,
-        boolean wasKilled, String diagonsticInfo,ReduceTaskStatus status) {
-      super(slotTaken, wasKilled, diagonsticInfo, status);
+    @Override
+    public TaskStatus getTaskStatus() {
+      return status;
+    }
+    
+    public void readFields(DataInput in) throws IOException {
+      super.readFields(in);
+      status = new ReduceTaskStatus();
+      status.readFields(in);
     }
   }
 

src/test/system/java/org/apache/hadoop/mapred/TestCluster.java

@@ -27,7 +27,9 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.examples.SleepJob;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.mapred.TaskStatus.State;
 import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.TaskID;
 import org.apache.hadoop.mapreduce.test.system.FinishTaskControlAction;
 import org.apache.hadoop.mapreduce.test.system.JTClient;
 import org.apache.hadoop.mapreduce.test.system.JTProtocol;
@@ -120,25 +122,18 @@ public class TestCluster {
   }
 
   /**
-   * Test to showcase how to get a task status from a TaskTracker.
-   * Does the following;
-   * 1. Contacts the job tracker to get TaskInfo
-   * 2. Uses taskinfo to get list of tts
-   * 3. Contacts TT and gets task info.
-   * 
-   * Care should be taken that the task which you are searching
-   * can need not be around.
+   * Test to verify the common properties of tasks.
    * @throws Exception
    */
   @Test
-  public void testTaskStatus() throws Exception {
+  public void testTaskDetails() throws Exception {
     Configuration conf = new Configuration(cluster.getConf());
     JTProtocol wovenClient = cluster.getJTClient().getProxy();
     FinishTaskControlAction.configureControlActionForJob(conf);
     SleepJob job = new SleepJob();
     job.setConf(conf);
 
-    conf = job.setupJobConf(1, 0, 100, 100, 100, 100);
+    conf = job.setupJobConf(1, 1, 100, 100, 100, 100);
     JobClient client = cluster.getJTClient().getClient();
 
     RunningJob rJob = client.submitJob(new JobConf(conf));
@@ -160,10 +155,22 @@ public class TestCluster {
         for(String taskTracker : taskTrackers) {
           TTInfo ttInfo = wovenClient.getTTInfo(taskTracker);
           TTClient ttCli =  cluster.getTTClient(ttInfo.getStatus().getHost());
-          TTTaskInfo ttTaskInfo = ttCli.getProxy().getTask(info.getTaskID());
+          TaskID taskId = info.getTaskID();
+          TTTaskInfo ttTaskInfo = ttCli.getProxy().getTask(taskId);
           Assert.assertNotNull(ttTaskInfo);
+          Assert.assertNotNull(ttTaskInfo.getConf());
+          Assert.assertNotNull(ttTaskInfo.getUser());
+          Assert.assertTrue(ttTaskInfo.getTaskStatus().getProgress() >= 0.0);
+          Assert.assertTrue(ttTaskInfo.getTaskStatus().getProgress() <= 1.0);
+          LOG.info("verified task progress to be between 0 and 1");
+          State state = ttTaskInfo.getTaskStatus().getRunState();
+          if (ttTaskInfo.getTaskStatus().getProgress() < 1.0 &&
+              ttTaskInfo.getTaskStatus().getProgress() >0.0) {
+            Assert.assertEquals(TaskStatus.State.RUNNING, state);
+            LOG.info("verified run state as " + state);
+          }
           FinishTaskControlAction action = new FinishTaskControlAction(
-              TaskID.downgrade(info.getTaskID()));
+              org.apache.hadoop.mapred.TaskID.downgrade(info.getTaskID()));
           ttCli.getProxy().sendAction(action);
         }
       }

src/test/system/java/org/apache/hadoop/mapreduce/test/system/TTTaskInfo.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.mapreduce.test.system;
 
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapred.TaskStatus;
 import org.apache.hadoop.mapred.TaskTracker;
@@ -27,13 +28,6 @@ import org.apache.hadoop.mapred.TaskTracker;
  */
 public interface TTTaskInfo extends Writable {
 
-  /**
-   * Gets the diagnostic information associated the the task.<br/>
-   * 
-   * @return diagnostic information of the task.
-   */
-  String getDiagnosticInfo();
-
   /**
    * Has task occupied a slot? A task occupies a slot once it starts localizing
    * on the {@link TaskTracker} <br/>
@@ -56,4 +50,22 @@ public interface TTTaskInfo extends Writable {
    * @return status of the particular task
    */
   TaskStatus getTaskStatus();
+  
+  /**
+   * Gets the configuration object of the task.
+   * @return
+   */
+  Configuration getConf();
+  
+  /**
+   * Gets the user of the task.
+   * @return
+   */
+  String getUser();
+  
+  /**
+   * Provides information as to whether the task is a cleanup of task.
+   * @return true if it is a clean up of task.
+   */
+  boolean isTaskCleanupTask();
 }