git commit: [SPARK-1620] Handle uncaught exceptions in function run by Akka scheduler
Repository: spark
Updated Branches:
refs/heads/branch-1.0 34f6fa921 -> 9ff9078fc
[SPARK-1620] Handle uncaught exceptions in function run by Akka scheduler
If the intended behavior was that uncaught exceptions thrown in functions being
run by the Akka scheduler would end up being handled by the default uncaught
exception handler set in Executor, and if that behavior is, in fact, correct,
then this is a way to accomplish that. I'm not certain, though, that we
shouldn't be doing something different to handle uncaught exceptions from some
of these scheduled functions.
In any event, this PR covers all of the cases I comment on in
[SPARK-1620](https://issues.apache.org/jira/browse/SPARK-1620).
Author: Mark Hamstra
Closes #622 from markhamstra/SPARK-1620 and squashes the following commits:
071d193 [Mark Hamstra] refactored post-SPARK-1772
1a6a35e [Mark Hamstra] another style fix
d30eb94 [Mark Hamstra] scalastyle
3573ecd [Mark Hamstra] Use wrapped try/catch in Utils.tryOrExit
8fc0439 [Mark Hamstra] Make functions run by the Akka scheduler use Executor's
UncaughtExceptionHandler
(cherry picked from commit 17f3075bc4aa8cbed165f7b367f70e84b1bc8db9)
Signed-off-by: Patrick Wendell
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/9ff9078f
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/9ff9078f
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/9ff9078f
Branch: refs/heads/branch-1.0
Commit: 9ff9078fc840c05c75f635b7a6acc5080b8e1185
Parents: 34f6fa9
Author: Mark Hamstra
Authored: Wed May 14 10:07:25 2014 -0700
Committer: Patrick Wendell
Committed: Wed May 14 10:07:39 2014 -0700
--
.../apache/spark/deploy/client/AppClient.scala| 18 ++
.../org/apache/spark/deploy/worker/Worker.scala | 18 ++
.../spark/scheduler/TaskSchedulerImpl.scala | 3 ++-
.../org/apache/spark/storage/BlockManager.scala | 2 +-
.../main/scala/org/apache/spark/util/Utils.scala | 13 +
5 files changed, 36 insertions(+), 18 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/9ff9078f/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
--
diff --git a/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
b/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
index 896913d..d38e9e7 100644
--- a/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
@@ -30,7 +30,7 @@ import org.apache.spark.{Logging, SparkConf, SparkException}
import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.master.Master
-import org.apache.spark.util.AkkaUtils
+import org.apache.spark.util.{Utils, AkkaUtils}
/**
* Interface allowing applications to speak with a Spark deploy cluster. Takes
a master URL,
@@ -88,13 +88,15 @@ private[spark] class AppClient(
var retries = 0
registrationRetryTimer = Some {
context.system.scheduler.schedule(REGISTRATION_TIMEOUT,
REGISTRATION_TIMEOUT) {
- retries += 1
- if (registered) {
-registrationRetryTimer.foreach(_.cancel())
- } else if (retries >= REGISTRATION_RETRIES) {
-markDead("All masters are unresponsive! Giving up.")
- } else {
-tryRegisterAllMasters()
+ Utils.tryOrExit {
+retries += 1
+if (registered) {
+ registrationRetryTimer.foreach(_.cancel())
+} else if (retries >= REGISTRATION_RETRIES) {
+ markDead("All masters are unresponsive! Giving up.")
+} else {
+ tryRegisterAllMasters()
+}
}
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/9ff9078f/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
--
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
index 85d25dc..134624c 100755
--- a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -166,14 +166,16 @@ private[spark] class Worker(
var retries = 0
registrationRetryTimer = Some {
context.system.scheduler.schedule(REGISTRATION_TIMEOUT,
REGISTRATION_TIMEOUT) {
-retries += 1
-if (registered) {
- registrationRetryTimer.foreach(_.cancel())
-} else if (retries >= REGISTRATION_RETRIES) {
- logError("All masters are unresponsive! Giving up.")
-
git commit: [SPARK-1620] Handle uncaught exceptions in function run by Akka scheduler
Repository: spark
Updated Branches:
refs/heads/master d58cb33ff -> 17f3075bc
[SPARK-1620] Handle uncaught exceptions in function run by Akka scheduler
If the intended behavior was that uncaught exceptions thrown in functions being
run by the Akka scheduler would end up being handled by the default uncaught
exception handler set in Executor, and if that behavior is, in fact, correct,
then this is a way to accomplish that. I'm not certain, though, that we
shouldn't be doing something different to handle uncaught exceptions from some
of these scheduled functions.
In any event, this PR covers all of the cases I comment on in
[SPARK-1620](https://issues.apache.org/jira/browse/SPARK-1620).
Author: Mark Hamstra
Closes #622 from markhamstra/SPARK-1620 and squashes the following commits:
071d193 [Mark Hamstra] refactored post-SPARK-1772
1a6a35e [Mark Hamstra] another style fix
d30eb94 [Mark Hamstra] scalastyle
3573ecd [Mark Hamstra] Use wrapped try/catch in Utils.tryOrExit
8fc0439 [Mark Hamstra] Make functions run by the Akka scheduler use Executor's
UncaughtExceptionHandler
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/17f3075b
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/17f3075b
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/17f3075b
Branch: refs/heads/master
Commit: 17f3075bc4aa8cbed165f7b367f70e84b1bc8db9
Parents: d58cb33
Author: Mark Hamstra
Authored: Wed May 14 10:07:25 2014 -0700
Committer: Patrick Wendell
Committed: Wed May 14 10:07:25 2014 -0700
--
.../apache/spark/deploy/client/AppClient.scala| 18 ++
.../org/apache/spark/deploy/worker/Worker.scala | 18 ++
.../spark/scheduler/TaskSchedulerImpl.scala | 3 ++-
.../org/apache/spark/storage/BlockManager.scala | 2 +-
.../main/scala/org/apache/spark/util/Utils.scala | 13 +
5 files changed, 36 insertions(+), 18 deletions(-)
--
http://git-wip-us.apache.org/repos/asf/spark/blob/17f3075b/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
--
diff --git a/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
b/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
index 896913d..d38e9e7 100644
--- a/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/client/AppClient.scala
@@ -30,7 +30,7 @@ import org.apache.spark.{Logging, SparkConf, SparkException}
import org.apache.spark.deploy.{ApplicationDescription, ExecutorState}
import org.apache.spark.deploy.DeployMessages._
import org.apache.spark.deploy.master.Master
-import org.apache.spark.util.AkkaUtils
+import org.apache.spark.util.{Utils, AkkaUtils}
/**
* Interface allowing applications to speak with a Spark deploy cluster. Takes
a master URL,
@@ -88,13 +88,15 @@ private[spark] class AppClient(
var retries = 0
registrationRetryTimer = Some {
context.system.scheduler.schedule(REGISTRATION_TIMEOUT,
REGISTRATION_TIMEOUT) {
- retries += 1
- if (registered) {
-registrationRetryTimer.foreach(_.cancel())
- } else if (retries >= REGISTRATION_RETRIES) {
-markDead("All masters are unresponsive! Giving up.")
- } else {
-tryRegisterAllMasters()
+ Utils.tryOrExit {
+retries += 1
+if (registered) {
+ registrationRetryTimer.foreach(_.cancel())
+} else if (retries >= REGISTRATION_RETRIES) {
+ markDead("All masters are unresponsive! Giving up.")
+} else {
+ tryRegisterAllMasters()
+}
}
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/17f3075b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
--
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
index 85d25dc..134624c 100755
--- a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -166,14 +166,16 @@ private[spark] class Worker(
var retries = 0
registrationRetryTimer = Some {
context.system.scheduler.schedule(REGISTRATION_TIMEOUT,
REGISTRATION_TIMEOUT) {
-retries += 1
-if (registered) {
- registrationRetryTimer.foreach(_.cancel())
-} else if (retries >= REGISTRATION_RETRIES) {
- logError("All masters are unresponsive! Giving up.")
- System.exit(1)
-} else {
- tryRegisterAllMasters()
+Utils.tryOrExit {
+
