Further address the comments

Change-Id: I63998f3eadcd32589b35d667ac10b65275a862eb

Author: jerryshao

core/src/main/scala/org/apache/spark/scheduler/BlacklistTracker.scala

@@ -151,7 +151,7 @@ private[scheduler] class BlacklistTracker (
       allocationClient match {
         case Some(a) =>
           logInfo(s"Killing blacklisted executor id $exec " +
-            s"since spark.blacklist.killBlacklistedExecutors is set.")
+            s"since ${config.BLACKLIST_KILL_ENABLED.key} is set.")
           a.killExecutors(Seq(exec), true, true)
         case None =>
           logWarning(s"Not attempting to kill blacklisted executor id $exec " +
@@ -165,7 +165,7 @@ private[scheduler] class BlacklistTracker (
       allocationClient match {
         case Some(a) =>
           logInfo(s"Killing all executors on blacklisted host $node " +
-            s"since spark.blacklist.killBlacklistedExecutors is set.")
+            s"since ${config.BLACKLIST_KILL_ENABLED.key} is set.")
           if (a.killExecutorsOnHost(node) == false) {
             logError(s"Killing executors on node $node failed.")
           }
@@ -179,7 +179,9 @@ private[scheduler] class BlacklistTracker (
   def updateBlacklistForFetchFailure(host: String, exec: String): Unit = {
     if (BLACKLIST_FETCH_FAILURE_ENABLED) {
       // If we blacklist on fetch failures, we are implicitly saying that we believe the failure is
-      // non-transient, and can't be recovered from (even if this is the first fetch failure).
+      // non-transient, and can't be recovered from (even if this is the first fetch failure,
+      // stage is retried after just one failure, so we don't always get a chance to collect
+      // multiple fetch failures).
       // If the external shuffle-service is on, then every other executor on this node would
       // be suffering from the same issue, so we should blacklist (and potentially kill) all
       // of them immediately.

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

@@ -51,31 +51,19 @@ import org.apache.spark.util.{AccumulatorV2, ThreadUtils, Utils}
  * acquire a lock on us, so we need to make sure that we don't try to lock the backend while
  * we are holding a lock on ourselves.
  */
-private[spark] class TaskSchedulerImpl private[scheduler](
+private[spark] class TaskSchedulerImpl(
     val sc: SparkContext,
     val maxTaskFailures: Int,
-    mockBlacklistTracker: Option[BlacklistTracker] = None,
     isLocal: Boolean = false)
   extends TaskScheduler with Logging {
 
   import TaskSchedulerImpl._
 
   def this(sc: SparkContext) = {
-    this(
-      sc,
-      sc.conf.get(config.MAX_TASK_FAILURES))
+    this(sc, sc.conf.get(config.MAX_TASK_FAILURES))
   }
 
-  def this(sc: SparkContext, maxTaskFailures: Int, isLocal: Boolean) = {
-    this(
-      sc,
-      maxTaskFailures,
-      mockBlacklistTracker = None,
-      isLocal = isLocal)
-  }
-
-  private[scheduler] lazy val blacklistTrackerOpt =
-    mockBlacklistTracker.orElse(maybeCreateBlacklistTracker(sc))
+  private[scheduler] lazy val blacklistTrackerOpt = maybeCreateBlacklistTracker(sc)
 
   val conf = sc.conf
 

core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala

@@ -87,7 +87,7 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with B
     conf.set(config.BLACKLIST_ENABLED, true)
     sc = new SparkContext(conf)
     taskScheduler =
-      new TaskSchedulerImpl(sc, sc.conf.getInt("spark.task.maxFailures", 4), Some(blacklist)) {
+      new TaskSchedulerImpl(sc, sc.conf.getInt("spark.task.maxFailures", 4)) {
         override def createTaskSetManager(taskSet: TaskSet, maxFailures: Int): TaskSetManager = {
           val tsm = super.createTaskSetManager(taskSet, maxFailures)
           // we need to create a spied tsm just so we can set the TaskSetBlacklist
@@ -98,6 +98,8 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext with B
           stageToMockTaskSetBlacklist(taskSet.stageId) = taskSetBlacklist
           tsmSpy
         }
+
+        override private[scheduler] lazy val blacklistTrackerOpt = Some(blacklist)
       }
     setupHelper()
   }