MAPREDUCE-7276 handle fast fail in MoveContainerToFailedFinishingTransition to fix fast fail not working issue

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java

@@ -2446,7 +2446,11 @@ public abstract class TaskAttemptImpl implements
       // register it to finishing state
       taskAttempt.appContext.getTaskAttemptFinishingMonitor().register(
           taskAttempt.attemptId);
-      notifyTaskAttemptFailed(taskAttempt, false);
+      boolean isFastFail = false;
+      if (event instanceof TaskAttemptFailEvent) {
+        isFastFail = ((TaskAttemptFailEvent) event).isFastFail();
+      }
+      notifyTaskAttemptFailed(taskAttempt, isFastFail);
     }
   }
 

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskAttempt.java

@@ -87,6 +87,7 @@ import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEventType;
+import org.apache.hadoop.mapreduce.v2.app.job.event.TaskTAttemptFailedEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskTAttemptKilledEvent;
 import org.apache.hadoop.mapreduce.v2.app.rm.ContainerRequestEvent;
 import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
@@ -1927,6 +1928,40 @@ public class TestTaskAttempt{
     assertFalse("InternalError occurred", eventHandler.internalError);
   }
 
+  @Test
+  public void testFastFailEnabledWhenFailFinishing() throws Exception {
+    MockEventHandler eventHandler = new MockEventHandler();
+    TaskAttemptImpl taImpl = createTaskAttemptImpl(eventHandler);
+    boolean isFastFail = true;
+
+    taImpl.handle(new TaskAttemptFailEvent(taImpl.getID(), isFastFail));
+
+    TaskEvent taskEvent = eventHandler.lastTaskEvent;
+    Assert.assertTrue("Task event is not an TaskTAttemptFailedEvent event",
+        taskEvent instanceof TaskTAttemptFailedEvent);
+    TaskTAttemptFailedEvent taskTAttemptFailedEvent =
+        (TaskTAttemptFailedEvent) taskEvent;
+    Assert.assertTrue("Fast fail is not true",
+        taskTAttemptFailedEvent.isFastFail());
+  }
+
+  @Test
+  public void testFastFailDisabledWhenFailFinishing() throws Exception {
+    MockEventHandler eventHandler = new MockEventHandler();
+    TaskAttemptImpl taImpl = createTaskAttemptImpl(eventHandler);
+    boolean isFastFail = false;
+
+    taImpl.handle(new TaskAttemptFailEvent(taImpl.getID(), isFastFail));
+
+    TaskEvent taskEvent = eventHandler.lastTaskEvent;
+    Assert.assertTrue("Task event is not an TaskTAttemptFailedEvent event",
+        taskEvent instanceof TaskTAttemptFailedEvent);
+    TaskTAttemptFailedEvent taskTAttemptFailedEvent =
+        (TaskTAttemptFailedEvent) taskEvent;
+    Assert.assertFalse("Fast fail is not false",
+        taskTAttemptFailedEvent.isFastFail());
+  }
+
   private void setupTaskAttemptFinishingMonitor(
       EventHandler eventHandler, JobConf jobConf, AppContext appCtx) {
     TaskAttemptFinishingMonitor taskAttemptFinishingMonitor =