You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by jo...@apache.org on 2015/12/02 22:44:09 UTC
spark git commit: [SPARK-12001] Allow partially-stopped
StreamingContext to be completely stopped
Repository: spark
Updated Branches:
refs/heads/master a1542ce2f -> 452690ba1
[SPARK-12001] Allow partially-stopped StreamingContext to be completely stopped
If `StreamingContext.stop()` is interrupted midway through the call, the context will be marked as stopped but certain state will have not been cleaned up. Because `state = STOPPED` will be set, subsequent `stop()` calls will be unable to finish stopping the context, preventing any new StreamingContexts from being created.
This patch addresses this issue by only marking the context as `STOPPED` once the `stop()` has successfully completed which allows `stop()` to be called a second time in order to finish stopping the context in case the original `stop()` call was interrupted.
I discovered this issue by examining logs from a failed Jenkins run in which this race condition occurred in `FailureSuite`, leaking an unstoppable context and causing all subsequent tests to fail.
Author: Josh Rosen <jo...@databricks.com>
Closes #9982 from JoshRosen/SPARK-12001.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/452690ba
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/452690ba
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/452690ba
Branch: refs/heads/master
Commit: 452690ba1cc3c667bdd9f3022c43c9a10267880b
Parents: a1542ce
Author: Josh Rosen <jo...@databricks.com>
Authored: Wed Dec 2 13:44:01 2015 -0800
Committer: Josh Rosen <jo...@databricks.com>
Committed: Wed Dec 2 13:44:01 2015 -0800
----------------------------------------------------------------------
.../spark/streaming/StreamingContext.scala | 49 +++++++++++---------
1 file changed, 27 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/452690ba/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
index 6fb8ad3..cf843e3 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala
@@ -699,28 +699,33 @@ class StreamingContext private[streaming] (
" AsynchronousListenerBus")
}
synchronized {
- try {
- state match {
- case INITIALIZED =>
- logWarning("StreamingContext has not been started yet")
- case STOPPED =>
- logWarning("StreamingContext has already been stopped")
- case ACTIVE =>
- scheduler.stop(stopGracefully)
- // Removing the streamingSource to de-register the metrics on stop()
- env.metricsSystem.removeSource(streamingSource)
- uiTab.foreach(_.detach())
- StreamingContext.setActiveContext(null)
- waiter.notifyStop()
- if (shutdownHookRef != null) {
- shutdownHookRefToRemove = shutdownHookRef
- shutdownHookRef = null
- }
- logInfo("StreamingContext stopped successfully")
- }
- } finally {
- // The state should always be Stopped after calling `stop()`, even if we haven't started yet
- state = STOPPED
+ // The state should always be Stopped after calling `stop()`, even if we haven't started yet
+ state match {
+ case INITIALIZED =>
+ logWarning("StreamingContext has not been started yet")
+ state = STOPPED
+ case STOPPED =>
+ logWarning("StreamingContext has already been stopped")
+ state = STOPPED
+ case ACTIVE =>
+ // It's important that we don't set state = STOPPED until the very end of this case,
+ // since we need to ensure that we're still able to call `stop()` to recover from
+ // a partially-stopped StreamingContext which resulted from this `stop()` call being
+ // interrupted. See SPARK-12001 for more details. Because the body of this case can be
+ // executed twice in the case of a partial stop, all methods called here need to be
+ // idempotent.
+ scheduler.stop(stopGracefully)
+ // Removing the streamingSource to de-register the metrics on stop()
+ env.metricsSystem.removeSource(streamingSource)
+ uiTab.foreach(_.detach())
+ StreamingContext.setActiveContext(null)
+ waiter.notifyStop()
+ if (shutdownHookRef != null) {
+ shutdownHookRefToRemove = shutdownHookRef
+ shutdownHookRef = null
+ }
+ logInfo("StreamingContext stopped successfully")
+ state = STOPPED
}
}
if (shutdownHookRefToRemove != null) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org