Skip to content

[SPARK-26891][YARN] Fixing flaky test in YarnSchedulerBackendSuite #23801

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 3 commits into from
Closed

[SPARK-26891][YARN] Fixing flaky test in YarnSchedulerBackendSuite #23801

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
90caf6d
initial commit
attilapiros Feb 15, 2019
765f7e8
Fixing resource leak
attilapiros Feb 16, 2019
1ff92bf
adding missing if
attilapiros Feb 17, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 29 additions & 8 deletions ...rs/yarn/src/test/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackendSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.spark.scheduler.cluster

import java.net.URL
import java.util.concurrent.atomic.AtomicReference
import javax.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse}

import scala.language.reflectiveCalls
Expand All @@ -32,15 +33,35 @@ import org.apache.spark.ui.TestFilter

class YarnSchedulerBackendSuite extends SparkFunSuite with MockitoSugar with LocalSparkContext {

private var yarnSchedulerBackend: YarnSchedulerBackend = _

override def afterEach() {
try {
if (yarnSchedulerBackend != null) {
yarnSchedulerBackend.stop()
}
} finally {
super.afterEach()
}
}

test("RequestExecutors reflects node blacklist and is serializable") {
sc = new SparkContext("local", "YarnSchedulerBackendSuite")
val sched = mock[TaskSchedulerImpl]
when(sched.sc).thenReturn(sc)
val yarnSchedulerBackend = new YarnSchedulerBackend(sched, sc) {
// Subclassing the TaskSchedulerImpl here instead of using Mockito. For details see SPARK-26891.
val sched = new TaskSchedulerImpl(sc) {
val blacklistedNodes = new AtomicReference[Set[String]]()

def setNodeBlacklist(nodeBlacklist: Set[String]): Unit = blacklistedNodes.set(nodeBlacklist)

override def nodeBlacklist(): Set[String] = blacklistedNodes.get()
}

val yarnSchedulerBackendExtended = new YarnSchedulerBackend(sched, sc) {
Copy link
Member

@srowen srowen Feb 17, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do you need the extra variable here, vs assigning to yarnSchedulerBackend? I don't see that they are used separately.

Copy link
Contributor Author

@attilapiros attilapiros Feb 17, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is needed because of the different type: the yarnSchedulerBackend type is YarnSchedulerBackend but yarnSchedulerBackendExtended type is an anonim subclass of YarnSchedulerBackend with the extra def setNodeBlacklist. On yarnSchedulerBackend I cannot call this extra method.

Copy link
Member

@srowen srowen Feb 17, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If so then how is it assigned in the next line? a subclass of YarnSchedulerBackend is still assignable to YarnSchedulerBackend. I might be missing something obvious here.

Copy link
Contributor Author

@attilapiros attilapiros Feb 17, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is assignable as yarnSchedulerBackendExtended is an instance of YarnSchedulerBackend too, although not a direct one.

def setHostToLocalTaskCount(hostToLocalTaskCount: Map[String, Int]): Unit = {
this.hostToLocalTaskCount = hostToLocalTaskCount
}
}
yarnSchedulerBackend = yarnSchedulerBackendExtended
val ser = new JavaSerializer(sc.conf).newInstance()
for {
blacklist <- IndexedSeq(Set[String](), Set("a", "b", "c"))
Expand All @@ -50,9 +71,9 @@ class YarnSchedulerBackendSuite extends SparkFunSuite with MockitoSugar with Loc
Map("a" -> 1, "b" -> 2)
)
} {
yarnSchedulerBackend.setHostToLocalTaskCount(hostToLocalCount)
when(sched.nodeBlacklist()).thenReturn(blacklist)
val req = yarnSchedulerBackend.prepareRequestExecutors(numRequested)
yarnSchedulerBackendExtended.setHostToLocalTaskCount(hostToLocalCount)
sched.setNodeBlacklist(blacklist)
val req = yarnSchedulerBackendExtended.prepareRequestExecutors(numRequested)
assert(req.requestedTotal === numRequested)
assert(req.nodeBlacklist === blacklist)
assert(req.hostToLocalTaskCount.keySet.intersect(blacklist).isEmpty)
Expand All @@ -75,9 +96,9 @@ class YarnSchedulerBackendSuite extends SparkFunSuite with MockitoSugar with Loc
// Before adding the "YARN" filter, should get the code from the filter in SparkConf.
assert(TestUtils.httpResponseCode(url) === HttpServletResponse.SC_BAD_GATEWAY)

val backend = new YarnSchedulerBackend(sched, sc) { }
yarnSchedulerBackend = new YarnSchedulerBackend(sched, sc) { }

backend.addWebUIFilter(classOf[TestFilter2].getName(),
yarnSchedulerBackend.addWebUIFilter(classOf[TestFilter2].getName(),
Map("responseCode" -> HttpServletResponse.SC_NOT_ACCEPTABLE.toString), "")

sc.ui.get.getHandlers.foreach { h =>
Expand Down