You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2015/04/13 13:21:35 UTC
spark git commit: [SPARK-6860][Streaming][WebUI] Fix the possible
inconsistency of StreamingPage
Repository: spark
Updated Branches:
refs/heads/master cadd7d72c -> 14ce3ea2c
[SPARK-6860][Streaming][WebUI] Fix the possible inconsistency of StreamingPage
Because `StreamingPage.render` doesn't hold the `listener` lock when generating the content, the different parts of content may have some inconsistent values if `listener` updates its status at the same time. And it will confuse people.
This PR added `listener.synchronized` to make sure we have a consistent view of StreamingJobProgressListener when creating the content.
Author: zsxwing <zs...@gmail.com>
Closes #5470 from zsxwing/SPARK-6860 and squashes the following commits:
cec6f92 [zsxwing] Add missing 'synchronized' in StreamingJobProgressListener
7182498 [zsxwing] Add synchronized to make sure we have a consistent view of StreamingJobProgressListener when creating the content
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/14ce3ea2
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/14ce3ea2
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/14ce3ea2
Branch: refs/heads/master
Commit: 14ce3ea2c9546c58203af85aceb76b1bfc1f650a
Parents: cadd7d7
Author: zsxwing <zs...@gmail.com>
Authored: Mon Apr 13 12:21:29 2015 +0100
Committer: Sean Owen <so...@cloudera.com>
Committed: Mon Apr 13 12:21:29 2015 +0100
----------------------------------------------------------------------
.../spark/streaming/ui/StreamingJobProgressListener.scala | 10 +++++-----
.../org/apache/spark/streaming/ui/StreamingPage.scala | 3 ++-
2 files changed, 7 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/14ce3ea2/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala
index 84f80e6..be1e868 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala
@@ -149,7 +149,7 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext)
}.toMap
}
- def lastReceivedBatchRecords: Map[Int, Long] = {
+ def lastReceivedBatchRecords: Map[Int, Long] = synchronized {
val lastReceivedBlockInfoOption = lastReceivedBatch.map(_.receivedBlockInfo)
lastReceivedBlockInfoOption.map { lastReceivedBlockInfo =>
(0 until numReceivers).map { receiverId =>
@@ -160,19 +160,19 @@ private[streaming] class StreamingJobProgressListener(ssc: StreamingContext)
}
}
- def receiverInfo(receiverId: Int): Option[ReceiverInfo] = {
+ def receiverInfo(receiverId: Int): Option[ReceiverInfo] = synchronized {
receiverInfos.get(receiverId)
}
- def lastCompletedBatch: Option[BatchInfo] = {
+ def lastCompletedBatch: Option[BatchInfo] = synchronized {
completedBatchInfos.sortBy(_.batchTime)(Time.ordering).lastOption
}
- def lastReceivedBatch: Option[BatchInfo] = {
+ def lastReceivedBatch: Option[BatchInfo] = synchronized {
retainedBatches.lastOption
}
- private def retainedBatches: Seq[BatchInfo] = synchronized {
+ private def retainedBatches: Seq[BatchInfo] = {
(waitingBatchInfos.values.toSeq ++
runningBatchInfos.values.toSeq ++ completedBatchInfos).sortBy(_.batchTime)(Time.ordering)
}
http://git-wip-us.apache.org/repos/asf/spark/blob/14ce3ea2/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
index bfe8086..b6dcb62 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingPage.scala
@@ -37,11 +37,12 @@ private[ui] class StreamingPage(parent: StreamingTab)
/** Render the page */
def render(request: HttpServletRequest): Seq[Node] = {
- val content =
+ val content = listener.synchronized {
generateBasicStats() ++ <br></br> ++
<h4>Statistics over last {listener.retainedCompletedBatches.size} processed batches</h4> ++
generateReceiverStats() ++
generateBatchStatsTable()
+ }
UIUtils.headerSparkPage("Streaming", content, parent, Some(5000))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org