You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by va...@apache.org on 2015/09/11 19:03:47 UTC
spark git commit: [SPARK-9924] [WEB UI] Don't schedule checkForLogs while some of them …
Repository: spark
Updated Branches:
refs/heads/branch-1.5 295281f52 -> 7f10bd620
[SPARK-9924] [WEB UI] Don't schedule checkForLogs while some of them …
Cherry-pick this to branch 1.5.
Author: Rohit Agarwal <ro...@qubole.com>
Closes #8701 from tgravescs/SPARK-9924-1.5 and squashes the following commits:
16e1c5f [Rohit Agarwal] [SPARK-9924] [WEB UI] Don't schedule checkForLogs while some of them are already running.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/7f10bd62
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/7f10bd62
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/7f10bd62
Branch: refs/heads/branch-1.5
Commit: 7f10bd620857aa558b77efcc4ba58380ea602f9b
Parents: 295281f
Author: Rohit Agarwal <ro...@qubole.com>
Authored: Fri Sep 11 10:03:39 2015 -0700
Committer: Marcelo Vanzin <va...@cloudera.com>
Committed: Fri Sep 11 10:03:39 2015 -0700
----------------------------------------------------------------------
.../deploy/history/FsHistoryProvider.scala | 28 +++++++++++++++-----
1 file changed, 21 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/7f10bd62/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
index 53c18ca..e573ff1 100644
--- a/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/history/FsHistoryProvider.scala
@@ -126,11 +126,11 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
// Disable the background thread during tests.
if (!conf.contains("spark.testing")) {
// A task that periodically checks for event log updates on disk.
- pool.scheduleAtFixedRate(getRunner(checkForLogs), 0, UPDATE_INTERVAL_S, TimeUnit.SECONDS)
+ pool.scheduleWithFixedDelay(getRunner(checkForLogs), 0, UPDATE_INTERVAL_S, TimeUnit.SECONDS)
if (conf.getBoolean("spark.history.fs.cleaner.enabled", false)) {
// A task that periodically cleans event logs on disk.
- pool.scheduleAtFixedRate(getRunner(cleanLogs), 0, CLEAN_INTERVAL_S, TimeUnit.SECONDS)
+ pool.scheduleWithFixedDelay(getRunner(cleanLogs), 0, CLEAN_INTERVAL_S, TimeUnit.SECONDS)
}
}
}
@@ -204,11 +204,25 @@ private[history] class FsHistoryProvider(conf: SparkConf, clock: Clock)
mod1 >= mod2
}
- logInfos.sliding(20, 20).foreach { batch =>
- replayExecutor.submit(new Runnable {
- override def run(): Unit = mergeApplicationListing(batch)
- })
- }
+ logInfos.grouped(20)
+ .map { batch =>
+ replayExecutor.submit(new Runnable {
+ override def run(): Unit = mergeApplicationListing(batch)
+ })
+ }
+ .foreach { task =>
+ try {
+ // Wait for all tasks to finish. This makes sure that checkForLogs
+ // is not scheduled again while some tasks are already running in
+ // the replayExecutor.
+ task.get()
+ } catch {
+ case e: InterruptedException =>
+ throw e
+ case e: Exception =>
+ logError("Exception while merging application listings", e)
+ }
+ }
lastModifiedTime = newLastModifiedTime
} catch {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org