MAPREDUCE-430. Reverting the patch.

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

CHANGES.txt

@@ -184,9 +184,6 @@ Release 0.20.1 - Unreleased
     MAPREDUCE-18. Puts some checks to detect cases where jetty serves up
     incorrect output during shuffle. (Ravi Gummadi via ddas)
 
-    MAPREDUCE-430. Fix bug related to Task getting stuck due to
-    OutOfMemoryErrors. (Amar Kamat via sharad)
-
 Release 0.20.0 - 2009-04-15
 
   INCOMPATIBLE CHANGES

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

@@ -180,13 +180,9 @@ class Child {
           break;
         }
       }
-    } catch (Error e) {
-      String error = "Error";
-      if (e instanceof FSError) {
-       error = "FSError";
-      }
-      LOG.fatal(error + " from child", e);
-      umbilical.taskError(taskid, e.getMessage());
+    } catch (FSError e) {
+      LOG.fatal("FSError from child", e);
+      umbilical.fsError(taskid, e.getMessage());
     } catch (Throwable throwable) {
       LOG.warn("Error running child", throwable);
       try {

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

@@ -51,9 +51,12 @@ public class IsolationRunner {
       LOG.info("Task " + taskid + " reporting done.");
     }
 
-    public void taskError(TaskAttemptID taskId, String message) 
-    throws IOException {
-      LOG.info("Task " + taskId + " reporting task error: " + message);
+    public void fsError(TaskAttemptID taskId, String message) throws IOException {
+      LOG.info("Task " + taskId + " reporting file system error: " + message);
+    }
+
+    public void shuffleError(TaskAttemptID taskId, String message) throws IOException {
+      LOG.info("Task " + taskId + " reporting shuffle error: " + message);
     }
 
     public JvmTask getTask(JVMId jvmId) throws IOException {

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

@@ -333,9 +333,13 @@ class LocalJobRunner implements JobSubmissionProtocol {
       }
     }
 
-    public void taskError(TaskAttemptID taskId, String message) 
+    public synchronized void fsError(TaskAttemptID taskId, String message) 
     throws IOException {
-      LOG.fatal("Error: "+ message + "from task: " + taskId);
+      LOG.fatal("FSError: "+ message + "from task: " + taskId);
+    }
+
+    public void shuffleError(TaskAttemptID taskId, String message) throws IOException {
+      LOG.fatal("shuffleError: "+ message + "from task: " + taskId);
     }
     
     public MapTaskCompletionEventsUpdate getMapCompletionEvents(JobID jobId, 

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

@@ -376,8 +376,8 @@ class ReduceTask extends Task {
         if(reduceCopier.mergeThrowable instanceof FSError) {
           LOG.error("Task: " + getTaskID() + " - FSError: " + 
               StringUtils.stringifyException(reduceCopier.mergeThrowable));
-          umbilical.taskError(getTaskID(), 
-              "(FSError) " + reduceCopier.mergeThrowable.getMessage());
+          umbilical.fsError(getTaskID(), 
+              reduceCopier.mergeThrowable.getMessage());
         }
         throw new IOException("Task: " + getTaskID() + 
             " - The reduce copier failed", reduceCopier.mergeThrowable);
@@ -1214,8 +1214,7 @@ class ReduceTask extends Task {
             LOG.error("Task: " + reduceTask.getTaskID() + " - FSError: " + 
                       StringUtils.stringifyException(e));
             try {
-              umbilical.taskError(reduceTask.getTaskID(), "(FSError) " 
-                                  + e.getMessage());
+              umbilical.fsError(reduceTask.getTaskID(), e.getMessage());
             } catch (IOException io) {
               LOG.error("Could not notify TT of FSError: " + 
                       StringUtils.stringifyException(io));
@@ -2092,9 +2091,9 @@ class ReduceTask extends Task {
                   LOG.fatal("Shuffle failed with too many fetch failures " + 
                             "and insufficient progress!" +
                             "Killing task " + getTaskID() + ".");
-                  umbilical.taskError(getTaskID(), "(Shuffle Error) " 
-                                      + "Exceeded MAX_FAILED_UNIQUE_FETCHES;"
-                                      + " bailing-out.");
+                  umbilical.shuffleError(getTaskID(), 
+                                         "Exceeded MAX_FAILED_UNIQUE_FETCHES;"
+                                         + " bailing-out.");
                 }
               }
                 

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

@@ -419,16 +419,12 @@ abstract class TaskRunner extends Thread {
               exitCode + ".");
         }
       }
-    } catch (Error e) {
-      String error = "Error";
-      if (e instanceof FSError) {
-        error = "FSError";
-      }
-      LOG.fatal(error, e);
+    } catch (FSError e) {
+      LOG.fatal("FSError", e);
       try {
-        tracker.taskError(t.getTaskID(), e.getMessage());
+        tracker.fsError(t.getTaskID(), e.getMessage());
       } catch (IOException ie) {
-        LOG.fatal(t.getTaskID()+" reporting " + error, ie);
+        LOG.fatal(t.getTaskID()+" reporting FSError", ie);
       }
     } catch (Throwable throwable) {
       LOG.warn(t.getTaskID()+" Child Error", throwable);

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

@@ -2569,14 +2569,25 @@ public class TaskTracker
   }
 
 
+  /** 
+   * A reduce-task failed to shuffle the map-outputs. Kill the task.
+   */  
+  public synchronized void shuffleError(TaskAttemptID taskId, String message) 
+  throws IOException { 
+    LOG.fatal("Task: " + taskId + " - Killed due to Shuffle Failure: " + message);
+    TaskInProgress tip = runningTasks.get(taskId);
+    tip.reportDiagnosticInfo("Shuffle Error: " + message);
+    purgeTask(tip, true);
+  }
+
   /** 
    * A child task had a local filesystem error. Kill the task.
    */  
-  public synchronized void taskError(TaskAttemptID taskId, String message) 
+  public synchronized void fsError(TaskAttemptID taskId, String message) 
   throws IOException {
-    LOG.fatal("Task: " + taskId + " - Killed due to : " + message);
+    LOG.fatal("Task: " + taskId + " - Killed due to FSError: " + message);
     TaskInProgress tip = runningTasks.get(taskId);
-    tip.reportDiagnosticInfo(message);
+    tip.reportDiagnosticInfo("FSError: " + message);
     purgeTask(tip, true);
   }
 

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

@@ -53,10 +53,9 @@ interface TaskUmbilicalProtocol extends VersionedProtocol {
    * Version 13 changed the getTask method signature for HADOOP-249
    * Version 14 changed the getTask method signature for HADOOP-4232
    * Version 15 Adds FAILED_UNCLEAN and KILLED_UNCLEAN states for HADOOP-4759
-   * Version 16 Removed fsError and shuffleError and introduced taskError.
    * */
 
-  public static final long versionID = 16L;
+  public static final long versionID = 15L;
   
   /**
    * Called when a child task process starts, to get its task.
@@ -123,8 +122,11 @@ interface TaskUmbilicalProtocol extends VersionedProtocol {
    */
   boolean canCommit(TaskAttemptID taskid) throws IOException;
 
-  /** Report that the task encountered an error.*/
-  void taskError(TaskAttemptID taskId, String message) throws IOException;
+  /** Report that a reduce-task couldn't shuffle map-outputs.*/
+  void shuffleError(TaskAttemptID taskId, String message) throws IOException;
+  
+  /** Report that the task encounted a local filesystem error.*/
+  void fsError(TaskAttemptID taskId, String message) throws IOException;
 
   /** Called by a reduce task to get the map output locations for finished maps.
    * Returns an update centered around the map-task-completion-events. 

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

@@ -49,9 +49,7 @@ public class TestTaskFail extends TestCase {
         throw new IOException();
       } else if (taskid.endsWith("_1")) {
         System.exit(-1);
-      } else if (taskid.endsWith("_2")) {
-        throw new OutOfMemoryError();
-      }
+      } 
     }
   }
 
@@ -108,57 +106,46 @@ public class TestTaskFail extends TestCase {
     return new JobClient(conf).submitJob(conf);
   }
   
-  private void validateAttempt(TaskInProgress tip, TaskAttemptID attemptId, 
-                               TaskStatus ts, boolean isCleanup) 
-  throws IOException {
-    assertEquals(tip.isCleanupAttempt(attemptId), isCleanup);
-    assertTrue(ts != null);
-    assertEquals(TaskStatus.State.FAILED, ts.getRunState());
-    // validate tasklogs for task attempt
-    String log = TestMiniMRMapRedDebugScript.readTaskLog(
-                      TaskLog.LogName.STDERR, attemptId, false);
-    assertTrue(log.contains(taskLog));
-    if (!isCleanup) {
-      // validate task logs: tasklog should contain both task logs
-      // and cleanup logs
-      assertTrue(log.contains(cleanupLog));
-    } else {
-      // validate tasklogs for cleanup attempt
-      log = TestMiniMRMapRedDebugScript.readTaskLog(
-                 TaskLog.LogName.STDERR, attemptId, true);
-      assertTrue(log.contains(cleanupLog));
-    }
-  }
-
   private void validateJob(RunningJob job, MiniMRCluster mr) 
   throws IOException {
     assertEquals(JobStatus.SUCCEEDED, job.getJobState());
 	    
     JobID jobId = job.getID();
     // construct the task id of first map task
-    // this should not be cleanup attempt since the first attempt 
-    // fails with an exception
     TaskAttemptID attemptId = 
       new TaskAttemptID(new TaskID(jobId, true, 0), 0);
     TaskInProgress tip = mr.getJobTrackerRunner().getJobTracker().
                             getTip(attemptId.getTaskID());
+    // this should not be cleanup attempt since the first attempt 
+    // fails with an exception
+    assertTrue(!tip.isCleanupAttempt(attemptId));
     TaskStatus ts = 
       mr.getJobTrackerRunner().getJobTracker().getTaskStatus(attemptId);
-    validateAttempt(tip, attemptId, ts, false);
+    assertTrue(ts != null);
+    assertEquals(TaskStatus.State.FAILED, ts.getRunState());
+    // validate task logs: tasklog should contain both task logs
+    // and cleanup logs
+    String log = TestMiniMRMapRedDebugScript.readTaskLog(
+                      TaskLog.LogName.STDERR, attemptId, false);
+    assertTrue(log.contains(taskLog));
+    assertTrue(log.contains(cleanupLog));
     
     attemptId =  new TaskAttemptID(new TaskID(jobId, true, 0), 1);
     // this should be cleanup attempt since the second attempt fails
     // with System.exit
-
+    assertTrue(tip.isCleanupAttempt(attemptId));
     ts = mr.getJobTrackerRunner().getJobTracker().getTaskStatus(attemptId);
+    assertTrue(ts != null);
+    assertEquals(TaskStatus.State.FAILED, ts.getRunState());
+    // validate tasklogs for task attempt
+    log = TestMiniMRMapRedDebugScript.readTaskLog(
+               TaskLog.LogName.STDERR, attemptId, false);
+    assertTrue(log.contains(taskLog));
 
-    validateAttempt(tip, attemptId, ts, true);
- 
-    attemptId =  new TaskAttemptID(new TaskID(jobId, true, 0), 2);
-    // this should be cleanup attempt since the third attempt fails
-    // with OutOfMemory
-    ts = mr.getJobTrackerRunner().getJobTracker().getTaskStatus(attemptId);
-    validateAttempt(tip, attemptId, ts, true);
+    // validate tasklogs for cleanup attempt
+    log = TestMiniMRMapRedDebugScript.readTaskLog(
+               TaskLog.LogName.STDERR, attemptId, true);
+    assertTrue(log.contains(cleanupLog));
   }
   
   public void testWithDFS() throws IOException {