Skip to content

[SPARK-24653][tests] Avoid cross-job pollution in TestUtils / SpillListener. #21639

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
vanzin wants to merge 2 commits into from
Closed

[SPARK-24653][tests] Avoid cross-job pollution in TestUtils / SpillListener. #21639

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9f0a9c4
[SPARK-24631][tests] Avoid cross-job pollution in TestUtils / SpillLi…
Jun 25, 2018
18d5ebf
Clean up return types.
Jun 26, 2018
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
Prev Previous commit
Clean up return types.
  • Loading branch information
Marcelo Vanzin committed Jun 26, 2018
commit 18d5ebfd201deaebf774835ec5eb08d2b6d08454
12 changes: 5 additions & 7 deletions core/src/main/scala/org/apache/spark/TestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,23 +172,21 @@ private[spark] object TestUtils {
/**
* Run some code involving jobs submitted to the given context and assert that the jobs spilled.
*/
def assertSpilled[T](sc: SparkContext, identifier: String)(body: => T): Unit = {
def assertSpilled(sc: SparkContext, identifier: String)(body: => Unit): Unit = {
withListener(sc, new SpillListener) { listener =>
val ret = body
body
assert(listener.numSpilledStages > 0, s"expected $identifier to spill, but did not")
ret
}
}

/**
* Run some code involving jobs submitted to the given context and assert that the jobs
* did not spill.
*/
def assertNotSpilled[T](sc: SparkContext, identifier: String)(body: => T): Unit = {
def assertNotSpilled(sc: SparkContext, identifier: String)(body: => Unit): Unit = {
withListener(sc, new SpillListener) { listener =>
val ret = body
body
assert(listener.numSpilledStages == 0, s"expected $identifier to not spill, but did")
ret
}
}

Expand Down Expand Up @@ -240,7 +238,7 @@ private[spark] object TestUtils {
* this method will wait until all events posted to the listener bus are processed, and then
* remove the listener from the bus.
*/
def withListener[L <: SparkListener, T](sc: SparkContext, listener: L) (body: L => T): T = {
def withListener[L <: SparkListener](sc: SparkContext, listener: L) (body: L => Unit): Unit = {
Copy link
Member

@srowen srowen Jul 31, 2018

Choose a reason for hiding this comment

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

private? hardly matters.

sc.addSparkListener(listener)
try {
body(listener)
Expand Down