SPARK-6735:[YARN] Adding properties to disable maximum number of executor failure's check or to make it relative to duration

yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala

@@ -59,6 +59,10 @@ private[spark] class ApplicationMaster(
   private val maxNumExecutorFailures = sparkConf.getInt("spark.yarn.max.executor.failures",
     sparkConf.getInt("spark.yarn.max.worker.failures", math.max(args.numExecutors * 2, 3)))
 
+  // Disable the maximum executor failure check
+  private val disableMaxExecutorFailureCheck = 
+              sparkConf.getBoolean("spark.yarn.max.executor.failures.disable", false)
+
   @volatile private var exitCode = 0
   @volatile private var unregistered = false
   @volatile private var finished = false
@@ -308,7 +312,8 @@ private[spark] class ApplicationMaster(
         var failureCount = 0
         while (!finished) {
           try {
-            if (allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
+            if (!disableMaxExecutorFailureCheck && 
+                 allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
               finish(FinalApplicationStatus.FAILED,
                 ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
                 "Max number of executor failures reached")

yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnAllocator.scala

@@ -20,6 +20,7 @@ package org.apache.spark.deploy.yarn
 import java.util.Collections
 import java.util.concurrent._
 import java.util.regex.Pattern
+import java.util.Stack
 
 import scala.collection.JavaConversions._
 import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet}
@@ -83,6 +84,9 @@ private[yarn] class YarnAllocator(
   private var executorIdCounter = 0
   @volatile private var numExecutorsFailed = 0
 
+  @volatile private var executorFailureTimeStamps = new Stack[Long]()
+  @volatile private var oldestRelativeExecutorFailure = -1L
+
   @volatile private var targetNumExecutors = args.numExecutors
 
   // Keep track of which container is running which executor to remove the executors later
@@ -94,6 +98,14 @@ private[yarn] class YarnAllocator(
   // Additional memory overhead.
   protected val memoryOverhead: Int = sparkConf.getInt("spark.yarn.executor.memoryOverhead",
     math.max((MEMORY_OVERHEAD_FACTOR * executorMemory).toInt, MEMORY_OVERHEAD_MIN))
+
+  // Make the maximum executor failure check to be relative with respect to duration
+  private val relativeMaxExecutorFailureCheck = 
+                sparkConf.getBoolean("spark.yarn.max.executor.failures.relative", false)
+  // window duration in sec ( default = 600 sec ) for checking maximum  number of executor failures
+  private val relativeMaxExecutorFailureCheckWindow = 
+               sparkConf.getInt("spark.yarn.max.executor.failures.relative.window", 600)
+
   // Number of cores per executor.
   protected val executorCores = args.executorCores
   // Resource capability requested for each executors
@@ -119,7 +131,27 @@ private[yarn] class YarnAllocator(
 
   def getNumExecutorsRunning: Int = numExecutorsRunning
 
-  def getNumExecutorsFailed: Int = numExecutorsFailed
+  def getNumExecutorsFailed: Int = {
+    if(relativeMaxExecutorFailureCheck){
+      getRelevantNumExecutorsFailed
+    } else {
+      numExecutorsFailed.intValue
+    }  
+  }
+
+  /**
+   *  Returns the the relative number of executor failures within the specifid window duration.
+   */
+
+  def getRelevantNumExecutorsFailed : Int = {
+    var currentTime = System.currentTimeMillis / 1000
+    var relevantWindowStartTime = currentTime - relativeMaxExecutorFailureCheckWindow
+    while(relevantWindowStartTime > oldestRelativeExecutorFailure && 
+          executorFailureTimeStamps.size > 0){
+      oldestRelativeExecutorFailure = executorFailureTimeStamps.pop
+    }
+    executorFailureTimeStamps.size + 1
+  }
 
   /**
    * Number of container requests that have not yet been fulfilled.
@@ -386,6 +418,9 @@ private[yarn] class YarnAllocator(
             ". Exit status: " + completedContainer.getExitStatus +
             ". Diagnostics: " + completedContainer.getDiagnostics)
           numExecutorsFailed += 1
+          if(relativeMaxExecutorFailureCheck) {
+             executorFailureTimeStamps.push(System.currentTimeMillis / 1000)
+          }
         }
       }