Github user squito commented on a diff in the pull request:
https://github.com/apache/spark/pull/8760#discussion_r42048350
--- Diff:
core/src/test/scala/org/apache/spark/scheduler/BlacklistTrackerSuite.scala ---
@@ -0,0 +1,162 @@
+package org.apache.spark.scheduler
+
+import org.apache.spark.SparkFunSuite
+import org.scalatest.BeforeAndAfter
+import org.apache.spark.SparkConf
+import org.apache.spark.TaskEndReason
+import org.apache.spark.Success
+import org.apache.spark.ExceptionFailure
+import org.apache.spark.LocalSparkContext
+import org.apache.spark.SparkContext
+
+import org.mockito.Mockito.{when,spy}
+
+class BlacklistTrackerSuite extends SparkFunSuite with BeforeAndAfter with
LocalSparkContext {
+
+ val FAILURE: TaskEndReason = new ExceptionFailure(
+ "Fake",
+ "fake failure",
+ Array.empty[StackTraceElement],
+ "fake stack trace",
+ None,
+ None)
+
+ // Variable name can indicate basic information of taskInfo
+ // The format is "taskInfo_executorId_taskId_hostName"
+ val taskInfo_1_1_hostA = new TaskInfo(1L,1,1,0L,"1","hostA",
TaskLocality.ANY, false)
+ val taskInfo_1_2_hostA = new TaskInfo(2L,1,1,0L,"1","hostA",
TaskLocality.ANY, false)
+ val taskInfo_2_1_hostA = new TaskInfo(1L,1,1,0L,"2","hostA",
TaskLocality.ANY, false)
+ val taskInfo_2_2_hostA = new TaskInfo(2L,1,1,0L,"2","hostA",
TaskLocality.ANY, false)
+ val taskInfo_3_3_hostB = new TaskInfo(3L,1,1,0L,"3","hostB",
TaskLocality.ANY, false)
+
+ test ("expireExecutorsInBlacklist works ") {
+ val conf = new SparkConf().setAppName("test").setMaster("local")
+ .set("spark.ui.enabled","false")
+ .set("spark.scheduler.blacklist.recoverPeriod", "2")
+ sc = new SparkContext(conf)
+ val scheduler = new TaskSchedulerImpl(sc, 1)
+
+ val tracker = new BlacklistTracker(conf)
+ tracker.start()
+ Thread.sleep(100)
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set("1"))
+
+ // sleep 2 second to wait for expireExecutorsInBlackList be triggered
+ Thread.sleep(2000)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set())
+
+ tracker.stop()
+ }
+
+ test("default strategy works as exceptation") {
+ val conf = new SparkConf().setAppName("test").setMaster("local")
+ .set("spark.ui.enabled","false")
+ sc = new SparkContext(conf)
+ val scheduler = new TaskSchedulerImpl(sc, 1)
+
+ val tracker = new BlacklistTracker(conf)
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, FAILURE)
+ tracker.updateFailureExecutors(taskInfo_2_1_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set("1","2"))
+ assert(tracker.executorBlacklist(scheduler, 2L) === Set())
+
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, Success)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set("2"))
+ assert(tracker.nodeBlacklist() === Set())
+
+ tracker.updateFailureExecutors(taskInfo_3_3_hostB, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 3L) === Set("3"))
+ assert(tracker.nodeBlacklist() === Set())
+
+ tracker.updateFailureExecutors(taskInfo_2_1_hostA, Success)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set())
+ }
+
+ test ("strict strategy works as exception") {
+ val conf = new SparkConf().setAppName("test").setMaster("local")
+ .set("spark.ui.enabled","false")
+ .set("spark.scheduler.blacklist.strategy", "strict")
+ sc = new SparkContext(conf)
+ val scheduler = new TaskSchedulerImpl(sc, 1)
+
+ val tracker = new BlacklistTracker(conf)
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set("1"))
+ tracker.updateFailureExecutors(taskInfo_1_2_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 2L) === Set("1"))
+ assert(tracker.nodeBlacklist() === Set("hostA"))
+
+ tracker.updateFailureExecutors(taskInfo_2_1_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 2L) === Set("1", "2"))
+ assert(tracker.nodeBlacklist() === Set("hostA"))
+
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, Success)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set("1","2"))
+
+ tracker.updateFailureExecutors(taskInfo_3_3_hostB, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 3L) === Set("1","2","3"))
+ assert(tracker.nodeBlacklist() === Set("hostA","hostB"))
+ }
+
+ test ("threshold strategy works as exception") {
+ val conf = new SparkConf().setAppName("test").setMaster("local")
+ .set("spark.ui.enabled","false")
+ .set("spark.scheduler.blacklist.strategy", "threshold")
+ .set("spark.scheduler.blacklist.threshold.maxFailureTaskNumber", "1")
+ .set("spark.scheduler.blacklist.threshold.maxBlackExecutorNumber",
"0")
+ sc = new SparkContext(conf)
+ val scheduler = new TaskSchedulerImpl(sc, 1)
+
+ val tracker = new BlacklistTracker(conf)
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set())
+ tracker.updateFailureExecutors(taskInfo_1_2_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 2L) === Set("1"))
+ assert(tracker.nodeBlacklist() === Set("hostA"))
+
+ tracker.updateFailureExecutors(taskInfo_2_1_hostA, FAILURE)
+ assert(tracker.executorBlacklist(scheduler, 2L) === Set("1"))
+ assert(tracker.nodeBlacklist() === Set("hostA"))
+
+ tracker.updateFailureExecutors(taskInfo_1_1_hostA, Success)
+ assert(tracker.executorBlacklist(scheduler, 1L) === Set())
--- End diff --
this last check is a little bit unusual. Is it really possible for this
task to get scheduled again on executor 1? It doesn't seem like there would be
any way for this task to get scheduled again on this host, before the blacklist
is updated. I suppose its possible, b/c the dagscheduler is told about the
task failure before the blacklist is updated, so it could get rescheduled on
the same executor before the blacklist is updated -- is that correct? But the
more "normal" route for an executor to get off the blacklist is just the
timeout.
its a good thing to test in any case, but I'd put in a comment explaining
how you'd end up in this situation.
You should also add a check that if a *different* task succeeds, the
executor stays in the blacklist.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
