Disabling the ping check feature by default and added a test case for ping check feature

Author: Ashutosh Gupta

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/TaskAttemptListenerImpl.java

@@ -410,7 +410,10 @@ public AMFeedback statusUpdate(TaskAttemptID taskAttemptID,
         LOG.debug("Ping from " + taskAttemptID.toString());
       }
       // Consider ping from the tasks for liveliness check
-      taskHeartbeatHandler.progressing(yarnAttemptID);
+      if (getConfig().getBoolean(MRJobConfig.MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK,
+          MRJobConfig.DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK)) {
+        taskHeartbeatHandler.progressing(yarnAttemptID);
+      }
       return feedback;
     }
 

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapred/TestTaskAttemptListenerImpl.java

@@ -415,18 +415,35 @@ public void testStatusUpdateProgress()
     startListener(true);
     verify(hbHandler).register(attemptId);
 
-    // make sure a ping does report progress
+    // make sure a ping doesn't report progress
     AMFeedback feedback = listener.statusUpdate(attemptID, null);
     assertTrue(feedback.getTaskFound());
-    verify(hbHandler, times(1)).progressing(eq(attemptId));
+    verify(hbHandler, never()).progressing(eq(attemptId));
 
     // make sure a status update does report progress
     MapTaskStatus mockStatus = new MapTaskStatus(attemptID, 0.0f, 1,
         TaskStatus.State.RUNNING, "", "RUNNING", "", TaskStatus.Phase.MAP,
         new Counters());
     feedback = listener.statusUpdate(attemptID, mockStatus);
     assertTrue(feedback.getTaskFound());
-    verify(hbHandler, times(2)).progressing(eq(attemptId));
+    verify(hbHandler).progressing(eq(attemptId));
+  }
+
+  @Test
+  public void testPingUpdateProgress() throws IOException, InterruptedException {
+    configureMocks();
+    Configuration conf = new Configuration();
+    conf.setBoolean(MRJobConfig.MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK, true);
+    listener.init(conf);
+    listener.start();
+    listener.registerPendingTask(task, wid);
+    listener.registerLaunchedTask(attemptId, wid);
+    verify(hbHandler).register(attemptId);
+
+    // make sure a ping does report progress
+    AMFeedback feedback = listener.statusUpdate(attemptID, null);
+    assertTrue(feedback.getTaskFound());
+    verify(hbHandler, times(1)).progressing(eq(attemptId));
   }
 
   @Test

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/MRJobConfig.java

@@ -919,6 +919,13 @@ public interface MRJobConfig {
     MR_AM_PREFIX + "scheduler.heartbeat.interval-ms";
   public static final int DEFAULT_MR_AM_TO_RM_HEARTBEAT_INTERVAL_MS = 1000;
 
+  /** Whether to consider ping from tasks in liveliness check. */
+  String MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK =
+      "mapreduce.task.enable.ping-for-liveliness-check";
+  boolean DEFAULT_MR_TASK_ENABLE_PING_FOR_LIVELINESS_CHECK
+      = false;
+
+
   /**
    * If contact with RM is lost, the AM will wait MR_AM_TO_RM_WAIT_INTERVAL_MS
    * milliseconds before aborting. During this interval, AM will still try

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml

@@ -286,6 +286,13 @@
   </description>
 </property>
 
+<property>
+  <name>mapreduce.task.enable.ping-for-liveliness-check</name>
+  <value>false</value>
+  <description>Whether to consider ping from tasks in liveliness check.
+  </description>
+</property>
+
 <property>
   <name>mapreduce.map.memory.mb</name>
   <value>-1</value>