commit 716e2e410d5f93da571ce1de3c84d38757fbe152

Author: Sharad Agarwal <sharadag@yahoo-inc.com>
Date:   Tue Feb 23 13:29:05 2010 +0530

     patch from


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

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

@@ -3,6 +3,8 @@ package org.apache.hadoop.mapred;
 import java.io.IOException;
 import java.util.List;
 import java.util.ArrayList;
+import java.util.Set;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapred.JobTracker.RetireJobInfo;
@@ -63,22 +65,7 @@ public privileged aspect JobTrackerAspect {
       LOG.warn("No task present for : " + taskID);
       return null;
     }
-    TaskInfo info;
-    TaskStatus[] status = tip.getTaskStatuses();
-    synchronized (tip) {
-      if (status == null) {
-        if (tip.isMapTask()) {
-          status = new MapTaskStatus[]{};
-        }
-        else {
-          status = new ReduceTaskStatus[]{};
-        }
-      }
-      info = new TaskInfoImpl(tip.getTIPId(), tip.getProgress(), tip
-          .getActiveTasks().size(), tip.numKilledTasks(), 
-          tip.numTaskFailures(), status);
-    }
-    return info;
+    return getTaskInfo(tip);
   }
 
   public TTInfo JobTracker.getTTInfo(String trackerName) throws IOException {
@@ -116,57 +103,18 @@ public privileged aspect JobTrackerAspect {
       return null;
     }
     List<TaskInfo> infoList = new ArrayList<TaskInfo>();
-    TaskStatus[] status;
     synchronized (jip) {
       for (TaskInProgress tip : jip.setup) {
-        status = tip.getTaskStatuses();
-        if (status == null) {
-          if (tip.isMapTask()) {
-            status = new MapTaskStatus[]{};
-          }
-          else {
-            status = new ReduceTaskStatus[]{};
-          }
-        }
-        TaskInfo info = new TaskInfoImpl(tip.getTIPId(), tip.getProgress(), tip
-            .getActiveTasks().size(), tip.numKilledTasks(), tip
-            .numTaskFailures(), status);
-        infoList.add(info);
+        infoList.add(getTaskInfo(tip));
       }
       for (TaskInProgress tip : jip.maps) {
-        status = tip.getTaskStatuses();
-        if (status == null) {
-          status = new MapTaskStatus[]{};
-        }
-        TaskInfo info = new TaskInfoImpl(tip.getTIPId(), tip.getProgress(), tip
-            .getActiveTasks().size(), tip.numKilledTasks(), tip
-            .numTaskFailures(), status);
-        infoList.add(info);
+        infoList.add(getTaskInfo(tip));
       }
       for (TaskInProgress tip : jip.reduces) {
-        status = tip.getTaskStatuses();
-        if (status == null) {
-          status = new ReduceTaskStatus[]{};
-        }
-        TaskInfo info = new TaskInfoImpl(tip.getTIPId(), tip.getProgress(), tip
-            .getActiveTasks().size(), tip.numKilledTasks(), tip
-            .numTaskFailures(), status);
-        infoList.add(info);
+        infoList.add(getTaskInfo(tip));
       }
       for (TaskInProgress tip : jip.cleanup) {
-        status = tip.getTaskStatuses();
-        if (status == null) {
-          if (tip.isMapTask()) {
-            status = new MapTaskStatus[]{};
-          }
-          else {
-            status = new ReduceTaskStatus[]{};
-          }
-        }
-        TaskInfo info = new TaskInfoImpl(tip.getTIPId(), tip.getProgress(), tip
-            .getActiveTasks().size(), tip.numKilledTasks(), tip
-            .numTaskFailures(), status);
-        infoList.add(info);
+        infoList.add(getTaskInfo(tip));
       }
     }
     return (TaskInfo[]) infoList.toArray(new TaskInfo[infoList.size()]);
@@ -231,4 +179,25 @@ public privileged aspect JobTrackerAspect {
         jobtrackerIndentifier) {
     tracker.setReady(true);
   }
+  
+  private TaskInfo JobTracker.getTaskInfo(TaskInProgress tip) {
+    TaskStatus[] status = tip.getTaskStatuses();
+    if (status == null) {
+      if (tip.isMapTask()) {
+        status = new MapTaskStatus[]{};
+      }
+      else {
+        status = new ReduceTaskStatus[]{};
+      }
+    }
+    String[] trackers =
+        (String[]) (tip.getActiveTasks().values()).toArray(new String[tip
+            .getActiveTasks().values().size()]);
+    TaskInfo info =
+        new TaskInfoImpl(tip.getTIPId(), tip.getProgress(), tip
+            .getActiveTasks().size(), tip.numKilledTasks(), tip
+            .numTaskFailures(), status, (tip.isJobSetupTask() || tip
+            .isJobCleanupTask()), trackers);
+    return info;
+  }
 }

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

@@ -30,20 +30,43 @@ public privileged aspect TaskTrackerAspect {
 
   public TTTaskInfo[] TaskTracker.getTasks() throws IOException {
     List<TTTaskInfo> infoList = new ArrayList<TTTaskInfo>();
-    for (TaskInProgress tip : tasks.values()) {
-      TTTaskInfo info = null;
-      if (tip.task.isMapTask()) {
-        info = new MapTTTaskInfo(((MapTask) tip.task), tip.slotTaken,
-            tip.wasKilled, tip.diagnosticInfo.toString());
-      } else {
-        info = new ReduceTTTaskInfo(((ReduceTask) tip.task), tip.slotTaken,
-            tip.wasKilled, tip.diagnosticInfo.toString());
+    synchronized (tasks) {
+      for (TaskInProgress tip : tasks.values()) {
+        TTTaskInfo info = getTTTaskInfo(tip);
+        infoList.add(info);
       }
-      infoList.add(info);
     }
     return (TTTaskInfo[]) infoList.toArray(new TTTaskInfo[infoList.size()]);
   }
 
+  public TTTaskInfo TaskTracker.getTask(org.apache.hadoop.mapreduce.TaskID id) 
+      throws IOException {
+    TaskID old = org.apache.hadoop.mapred.TaskID.downgrade(id);
+    synchronized (tasks) {
+      for(TaskAttemptID ta : tasks.keySet()) {
+        if(old.equals(ta.getTaskID())) {
+          return getTTTaskInfo(tasks.get(ta));
+        }
+      }
+    }
+    return null;
+  }
+
+  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());
+    } else {
+      info =
+          new ReduceTTTaskInfo(
+              tip.slotTaken, tip.wasKilled, tip.diagnosticInfo.toString(),
+              (ReduceTaskStatus) tip.getStatus());
+    }
+    return info;
+  }
+
   before(TaskTrackerStatus newStatus, TaskTracker tracker) : 
     set(TaskTrackerStatus TaskTracker.status) 
     && args(newStatus) && this(tracker) {

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

@@ -16,20 +16,20 @@ import org.apache.hadoop.mapreduce.test.system.TTTaskInfo;
 abstract class TTTaskInfoImpl implements TTTaskInfo {
 
   private String diagonsticInfo;
-  private Task task;
   private boolean slotTaken;
   private boolean wasKilled;
+  TaskStatus status;
 
   public TTTaskInfoImpl() {
   }
 
-  public TTTaskInfoImpl(Task task, boolean slotTaken, boolean wasKilled,
-      String diagonsticInfo) {
+  public TTTaskInfoImpl(boolean slotTaken, boolean wasKilled,
+      String diagonsticInfo, TaskStatus status) {
     super();
     this.diagonsticInfo = diagonsticInfo;
-    this.task = task;
     this.slotTaken = slotTaken;
     this.wasKilled = wasKilled;
+    this.status = status;
   }
 
   @Override
@@ -37,11 +37,6 @@ abstract class TTTaskInfoImpl implements TTTaskInfo {
     return diagonsticInfo;
   }
 
-  @Override
-  public Task getTask() {
-    return task;
-  }
-
   @Override
   public boolean slotTaken() {
     return slotTaken;
@@ -52,45 +47,49 @@ abstract class TTTaskInfoImpl implements TTTaskInfo {
     return wasKilled;
   }
 
+  @Override
+  public TaskStatus getTaskStatus() {
+    return status;
+  }
+
   @Override
   public void readFields(DataInput in) throws IOException {
-    task.readFields(in);
     slotTaken = in.readBoolean();
     wasKilled = in.readBoolean();
     diagonsticInfo = in.readUTF();
+    status.readFields(in);
   }
 
   @Override
   public void write(DataOutput out) throws IOException {
-    task.write(out);
     out.writeBoolean(slotTaken);
     out.writeBoolean(wasKilled);
     out.writeUTF(diagonsticInfo);
+    status.write(out);
   }
 
   static class MapTTTaskInfo extends TTTaskInfoImpl {
 
     public MapTTTaskInfo() {
-      super(new MapTask(), false, false, "");
+      super(false, false, "", new MapTaskStatus());
     }
 
-    public MapTTTaskInfo(MapTask task, boolean slotTaken, boolean wasKilled,
-        String diagonsticInfo) {
-      super(task, slotTaken, wasKilled, diagonsticInfo);
+    public MapTTTaskInfo(boolean slotTaken, boolean wasKilled,
+        String diagonsticInfo, MapTaskStatus status) {
+      super(slotTaken, wasKilled, diagonsticInfo, status);
     }
   }
 
   static class ReduceTTTaskInfo extends TTTaskInfoImpl {
 
     public ReduceTTTaskInfo() {
-      super(new ReduceTask(), false, false, "");
+      super(false, false, "", new ReduceTaskStatus());
     }
 
-    public ReduceTTTaskInfo(ReduceTask task, boolean slotTaken,
-        boolean wasKilled, String diagonsticInfo) {
-      super(task, slotTaken, wasKilled, diagonsticInfo);
+    public ReduceTTTaskInfo(boolean slotTaken,
+        boolean wasKilled, String diagonsticInfo,ReduceTaskStatus status) {
+      super(slotTaken, wasKilled, diagonsticInfo, status);
     }
-
   }
 
 }

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

@@ -21,12 +21,17 @@ class TaskInfoImpl implements TaskInfo {
   private int failedAttempts;
   private int runningAttempts;
   private TaskStatus[] taskStatus;
+  private boolean setupOrCleanup;
+  private String[] taskTrackers;
 
   public TaskInfoImpl() {
     taskID = new TaskID();
   }
-  public TaskInfoImpl(TaskID taskID, double progress, int runningAttempts,
-      int killedAttempts, int failedAttempts, TaskStatus[] taskStatus) {
+
+  public TaskInfoImpl(
+      TaskID taskID, double progress, int runningAttempts, int killedAttempts,
+      int failedAttempts, TaskStatus[] taskStatus,
+      boolean setupOrCleanup, String[] taskTrackers) {
     this.progress = progress;
     this.taskID = taskID;
     this.killedAttempts = killedAttempts;
@@ -34,16 +39,15 @@ class TaskInfoImpl implements TaskInfo {
     this.runningAttempts = runningAttempts;
     if (taskStatus != null) {
       this.taskStatus = taskStatus;
-    }
-    else { 
+    } else {
       if (taskID.isMap()) {
-        this.taskStatus = new MapTaskStatus[]{};
-      }
-      else {
-        this.taskStatus = new ReduceTaskStatus[]{};
+        this.taskStatus = new MapTaskStatus[] {};
+      } else {
+        this.taskStatus = new ReduceTaskStatus[] {};
       }
     }
-    
+    this.setupOrCleanup = setupOrCleanup;
+    this.taskTrackers = taskTrackers;
   }
 
   @Override
@@ -93,6 +97,13 @@ class TaskInfoImpl implements TaskInfo {
         taskStatus[i] = new ReduceTaskStatus();
       }
       taskStatus[i].readFields(in);
+      taskStatus[i].setTaskTracker(in.readUTF());
+    }
+    setupOrCleanup = in.readBoolean();
+    size = in.readInt();
+    taskTrackers = new String[size];
+    for(int i=0; i < size ; i++) {
+      taskTrackers[i] = in.readUTF();
     }
   }
 
@@ -106,11 +117,27 @@ class TaskInfoImpl implements TaskInfo {
     out.writeInt(taskStatus.length);
     for (TaskStatus t : taskStatus) {
       t.write(out);
+      out.writeUTF(t.getTaskTracker());
+    }
+    out.writeBoolean(setupOrCleanup);
+    out.writeInt(taskTrackers.length);
+    for(String tt : taskTrackers) {
+      out.writeUTF(tt);
     }
   }
-  
+
   @Override
   public TaskStatus[] getTaskStatus() {
     return taskStatus;
   }
+
+  @Override
+  public boolean isSetupOrCleanup() {
+    return setupOrCleanup;
+  }
+
+  @Override
+  public String[] getTaskTrackers() {
+    return taskTrackers;
+  }
 }

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

@@ -9,11 +9,17 @@ 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.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.test.system.FinishTaskControlAction;
 import org.apache.hadoop.mapreduce.test.system.JTClient;
 import org.apache.hadoop.mapreduce.test.system.JTProtocol;
 import org.apache.hadoop.mapreduce.test.system.JobInfo;
 import org.apache.hadoop.mapreduce.test.system.MRCluster;
 import org.apache.hadoop.mapreduce.test.system.TTClient;
+import org.apache.hadoop.mapreduce.test.system.TTInfo;
+import org.apache.hadoop.mapreduce.test.system.TTTaskInfo;
+import org.apache.hadoop.mapreduce.test.system.TaskInfo;
+import org.apache.hadoop.net.NetUtils;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -98,4 +104,56 @@ public class TestCluster {
         +"  " + fz.getClass());
   }
 
+  /**
+   * 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.
+   * @throws Exception
+   */
+  @Test
+  public void testTaskStatus() throws Exception {
+    Configuration conf = new Configuration(cluster.getConf());
+    JTProtocol wovenClient = cluster.getMaster().getProxy();
+    FinishTaskControlAction.configureControlActionForJob(conf);
+    SleepJob job = new SleepJob();
+    job.setConf(conf);
+
+    conf = job.setupJobConf(1, 0, 100, 100, 100, 100);
+    JobClient client = cluster.getMaster().getClient();
+
+    RunningJob rJob = client.submitJob(new JobConf(conf));
+    JobID id = rJob.getID();
+
+    JobInfo jInfo = wovenClient.getJobInfo(id);
+
+    while (jInfo.getStatus().getRunState() != JobStatus.RUNNING) {
+      Thread.sleep(1000);
+      jInfo = wovenClient.getJobInfo(id);
+    }
+
+    LOG.info("Waiting till job starts running one map");
+
+    TaskInfo[] myTaskInfos = wovenClient.getTaskInfo(id);
+    for(TaskInfo info : myTaskInfos) {
+      if(!info.isSetupOrCleanup()) {
+        String[] taskTrackers = info.getTaskTrackers();
+        for(String taskTracker : taskTrackers) {
+          TTInfo ttInfo = wovenClient.getTTInfo(taskTracker);
+          TTClient ttCli =  cluster.getSlaves().get(
+              ttInfo.getStatus().getHost());
+          TTTaskInfo ttTaskInfo = ttCli.getProxy().getTask(info.getTaskID());
+          Assert.assertNotNull(ttTaskInfo);
+          FinishTaskControlAction action = new FinishTaskControlAction(
+              TaskID.downgrade(info.getTaskID()));
+          ttCli.getProxy().sendAction(action);
+        }
+      }
+    }
+    rJob.killJob();
+  }
 }

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

@@ -5,6 +5,7 @@ import java.io.IOException;
 import org.apache.hadoop.mapred.JobTracker;
 import org.apache.hadoop.mapred.TaskTracker;
 import org.apache.hadoop.mapred.TaskTrackerStatus;
+import org.apache.hadoop.mapreduce.TaskID;
 import org.apache.hadoop.test.system.DaemonProtocol;
 
 /**
@@ -29,4 +30,14 @@ public interface TTProtocol extends DaemonProtocol {
    * @throws IOException
    */
   TTTaskInfo[] getTasks() throws IOException;
+
+  /**
+   * Gets the task associated with the id.<br/>
+   * 
+   * @param id of the task.
+   * 
+   * @return
+   * @throws IOException
+   */
+  TTTaskInfo getTask(TaskID taskID) throws IOException;
 }

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

@@ -1,20 +1,13 @@
 package org.apache.hadoop.mapreduce.test.system;
 
 import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.mapred.Task;
+import org.apache.hadoop.mapred.TaskStatus;
 import org.apache.hadoop.mapred.TaskTracker;
 
 /**
  * Task state information as seen by the TT.
  */
 public interface TTTaskInfo extends Writable {
-  /**
-   * Gets the task associated to the instance as seen by {@link TaskTracker}
-   * <br/>
-   * 
-   * @return task.
-   */
-  Task getTask();
 
   /**
    * Gets the diagnostic information associated the the task.<br/>
@@ -37,4 +30,12 @@ public interface TTTaskInfo extends Writable {
    * @return true, if task has been killed.
    */
   boolean wasKilled();
+
+  /**
+   * Gets the task status associated with the particular task trackers task 
+   * view.<br/>
+   * 
+   * @return status of the particular task
+   */
+  TaskStatus getTaskStatus();
 }

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

@@ -49,9 +49,25 @@ public interface TaskInfo extends Writable {
 
   /**
    * Array of TaskStatus objects that are related to the corresponding
-   * TaskInProgress object.
+   * TaskInProgress object.The task status of the tip is only populated
+   * once a tracker reports back the task status.<br/>
    * 
-   * @return
+   * @return list of task statuses.
    */
   TaskStatus[] getTaskStatus();
+
+  /**
+   * Gets a list of tracker on which the task attempts are scheduled/running.
+   * Can be empty if the task attempt has succeeded <br/>
+   * 
+   * @return list of trackers
+   */
+  String[] getTaskTrackers();
+
+  /**
+   * Gets if the current TaskInProgress is a setup or cleanup tip. <br/>
+   * 
+   * @return true if setup/cleanup
+   */
+  boolean isSetupOrCleanup();
 }

src/test/system/java/org/apache/hadoop/test/system/process/HadoopDaemonRemoteCluster.java

@@ -4,6 +4,7 @@ import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
+import java.net.InetAddress;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Map;
@@ -236,8 +237,10 @@ public class HadoopDaemonRemoteCluster implements ClusterProcessManager {
       reader = new BufferedReader(new FileReader(slavesFile));
       String slaveHost = null;
       while ((slaveHost = reader.readLine()) != null) {
-        RemoteProcess slave = new ScriptDaemon(slaveCommand, slaveHost);
-        slaves.put(slaveHost, slave);
+        InetAddress addr = InetAddress.getByName(slaveHost);
+        RemoteProcess slave = new ScriptDaemon(slaveCommand, 
+            addr.getCanonicalHostName());
+        slaves.put(addr.getCanonicalHostName(), slave);
       }
       if (slaves.size() == 0) {
         LOG.error(confLocation + System.getProperty("file.separator", "/") +