-
Notifications
You must be signed in to change notification settings - Fork 28.3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
SPARK-1407 drain event queue before stopping event logger
Author: Kan Zhang <kzhang@apache.org> Closes #366 from kanzhang/SPARK-1407 and squashes the following commits: cd0629f [Kan Zhang] code refactoring and adding test b073ee6 [Kan Zhang] SPARK-1407 drain event queue before stopping event logger
- Loading branch information
Showing
4 changed files
with
67 additions
and
16 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -17,6 +17,8 @@ | |
|
||
package org.apache.spark.scheduler | ||
|
||
import java.util.concurrent.Semaphore | ||
|
||
import scala.collection.mutable | ||
|
||
import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSuite} | ||
|
@@ -72,6 +74,49 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc | |
} | ||
} | ||
|
||
test("bus.stop() waits for the event queue to completely drain") { | ||
@volatile var drained = false | ||
|
||
// Tells the listener to stop blocking | ||
val listenerWait = new Semaphore(1) | ||
|
||
// When stop has returned | ||
val stopReturned = new Semaphore(1) | ||
|
||
class BlockingListener extends SparkListener { | ||
override def onJobEnd(jobEnd: SparkListenerJobEnd) = { | ||
listenerWait.acquire() | ||
drained = true | ||
} | ||
} | ||
|
||
val bus = new LiveListenerBus | ||
val blockingListener = new BlockingListener | ||
|
||
bus.addListener(blockingListener) | ||
bus.start() | ||
bus.post(SparkListenerJobEnd(0, JobSucceeded)) | ||
|
||
// the queue should not drain immediately | ||
assert(!drained) | ||
|
||
new Thread("ListenerBusStopper") { | ||
override def run() { | ||
// stop() will block until notify() is called below | ||
bus.stop() | ||
stopReturned.release(1) | ||
} | ||
}.start() | ||
|
||
while (!bus.stopCalled) { | ||
Thread.sleep(10) | ||
} | ||
|
||
listenerWait.release() | ||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong.
concretevitamin
Contributor
|
||
stopReturned.acquire() | ||
assert(drained) | ||
} | ||
|
||
test("basic creation of StageInfo") { | ||
val listener = new SaveStageAndTaskInfo | ||
sc.addSparkListener(listener) | ||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -73,6 +73,6 @@ object SparkHdfsLR { | |
} | ||
|
||
println("Final w: " + w) | ||
System.exit(0) | ||
sc.stop() | ||
} | ||
} |
Shouldn't we assert(!drained) after stop() is called but before listenerWait.release(), to make sure listenerThread is still blocked? I was thinking along the same line, but wasn't sure a single assert statement is enough to tell listenerThread is blocked (maybe it didn't get chance to run).