Duplicate (?) code paths to handle Executor failures

[Kay Ousterhout <ke...@eecs.berkeley.edu>]

Hi all,

I noticed that when the JVM for an executor fails, in Standalone mode, we
have two duplicate code paths that handle the failure, one via Akka, and
the second via the Worker/ExecutorRunner:

via Akka:
(1) CoarseGrainedSchedulerBackend is notified that the remote Akka endpoint
is disconnected:
https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala#L189
and it calls CoarseGrainedSchedulerBackend.removeExecutor
(2) removeExecutor() tells the task scheduler to reschedule all of the
tasks that were running on that executor

via the Worker/ExecutorRunner:
(1) The ExecutorRunner notes that the Executor process has failed and
notifies the Spark Master
(2) The Master notifies the AppClient (for the particular application),
which then notifies the SparkDeploySchedulerBackend (which subclasses
SparkDeploySchedulerBackend)
(3) SparkDeploySchedulerBackend calls
CoarseGrainedSchedulerBackend.removeExecutor, which eventually tells the
task scheduler that the executor was lost and all tasks running on it
should be re-scheduled (as above)

For YARN, my understanding is that there is a 3rd code path where the
YarnAllocator's processCompletedContainers() gets information about the
process's exit from the master, and translates it into an "ExecutorExited"
message that gets passed to the scheduler, similar to in the
Worker/ExecutorRunner case (YARN folks, is this correct?).

It's confusing and error prone to have these multiple different ways of
handling failures (I ran into this problem because I was fixing a bug where
one of the code paths can lead to a hang, but the other one doesn't).  Can
we eliminate all but one of these code paths?  Is there a reason for the
duplicate error handling?

Do all of the cluster managers (Standalone, YARN, Mesos) communicate in
some way when an Executor has failed, so we can ignore the Akka code path?
The Akka code path is most tempting to eliminate because it has less
information about the failure (the other code path typically has an exit
code for the process, at a minimum).

I'm also curious if others have seen this issue; for example, Marcelo, I'm
wondering if this came up in your attempts to treat YARN pre-emption
differently (did you run into issues where, when YARN pre-empts an
executor, Spark gets the "Rpc disassociated" failure from AKKA before the
more useful error from Yarn saying that the executor was pre-empted?).

-Kay

----------------------------------

To reproduce this issue, you can run one of these jobs:

sc.parallelize(1 to 10, 2).foreach { x => if (x == 1) throw new
OutOfMemoryError("test OOM") }

or

sc.parallelize(1 to 10, 2).foreach { x => if (x == 1) System.exit(42) }