apache/incubator-spark — pull request #440 by CodingCat (2014-01-15 17:53:39+00:00)
Title: the failure start of executor should not restart the worker, otherwise there would be an cascading failure

the executor will be failed to start due to the user-end bug

e.g.

if the user end didn't set SPARK_HOME, the following line in Worker.scala will throw an null pointer exception (because you pass a NULL to new File)

val manager = new ExecutorRunner(appId, execId, appDesc, cores_, memory_,
self, workerId, host, new File(execSparkHome_), workDir, akkaUrl, ExecutorState.RUNNING)

this exception will make Worker actor restart (default behaviour of akka actor), registering itself with a new workerID (but same worker address), because of

def generateWorkerId(): String = {
"worker-%s-%s-%d".format(DATE_FORMAT.format(new Date), host, port)
}

considering the following code in Master.scala

case RegisterWorker(id, workerHost, workerPort, cores, memory, workerWebUiPort, publicAddress) => {
logInfo("Registering worker %s:%d with %d cores, %s RAM".format(
host, workerPort, cores, Utils.megabytesToString(memory)))
if (state == RecoveryState.STANDBY) {
// ignore, don't send response
} else if (idToWorker.contains(id)) {
sender ! RegisterWorkerFailed("Duplicate worker ID")
} else {
val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
sender, workerWebUiPort, publicAddress)
registerWorker(worker)
persistenceEngine.addWorker(worker)
sender ! RegisteredWorker(masterUrl, masterWebUiUrl)
schedule()
}
}

the Master will go into registerWorker and found that the address is already there

def registerWorker(worker: WorkerInfo): Unit = {
// There may be one or more refs to dead workers on this same node (w/ different ID's),
// remove them.
workers.filter { w =>
(w.host == host && w.port == port) && (w.state == WorkerState.DEAD)
}.foreach { w =>
workers -= w
}

val workerAddress = worker.actor.path.address
if (addressToWorker.contains(workerAddress)) {
logInfo("Attempted to re-register worker at same address: " + workerAddress)
return
}

workers += worker
idToWorker(worker.id) = worker <- THIS LINE WILL NOT BE EXECUTED
actorToWorker(worker.actor) = worker
addressToWorker(workerAddress) = worker
}

so in the future time, master will always warn that "receive heartbeat from unregistered worker xxx" where xxx is the new workerID generated when restarting the worker,

after some time, the user will see the DEAD workers in the webUI and the application is always in LAUNCHING status

MY FIX:

capture exception when starting the executor;

send the failed status to master if any exception happens