SPARK-2425 introduce LOADING -> RUNNING ApplicationState transition

and prevent Master from removing Application with RUNNING Executors

Conflicts:
	core/src/main/scala/org/apache/spark/deploy/master/Master.scala
	core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala
	core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

Author: markhamstra

core/src/main/scala/org/apache/spark/deploy/master/ApplicationInfo.scala

@@ -90,11 +90,13 @@ private[spark] class ApplicationInfo(
 
   def retryCount = _retryCount
 
-  def incrementRetryCount = {
+  def incrementRetryCount() = {
     _retryCount += 1
     _retryCount
   }
 
+  def resetRetryCount() = _retryCount = 0
+
   def markFinished(endState: ApplicationState.Value) {
     state = endState
     endTime = System.currentTimeMillis()

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

@@ -264,22 +264,29 @@ private[spark] class Master(host: String, port: Int, webUiPort: Int) extends Act
       val execOption = idToApp.get(appId).flatMap(app => app.executors.get(execId))
       execOption match {
         case Some(exec) => {
+          val appInfo = idToApp(appId)
           exec.state = state
+          if (state == ExecutorState.RUNNING) { appInfo.resetRetryCount() }
           exec.application.driver ! ExecutorUpdated(execId, state, message, exitStatus)
           if (ExecutorState.isFinished(state)) {
-            val appInfo = idToApp(appId)
             // Remove this executor from the worker and app
             logInfo("Removing executor " + exec.fullId + " because it is " + state)
             appInfo.removeExecutor(exec)
             exec.worker.removeExecutor(exec)
 
+            val normalExit = exitStatus == Some(0)
             // Only retry certain number of times so we don't go into an infinite loop.
-            if (appInfo.incrementRetryCount < ApplicationState.MAX_NUM_RETRY) {
-              schedule()
-            } else {
-              logError("Application %s with ID %s failed %d times, removing it".format(
-                appInfo.desc.name, appInfo.id, appInfo.retryCount))
-              removeApplication(appInfo, ApplicationState.FAILED)
+            if (!normalExit) {
+              if (appInfo.incrementRetryCount() < ApplicationState.MAX_NUM_RETRY) {
+                schedule()
+              } else {
+                logError("Application %s with ID %s failed %d times, removing it".format(
+                  appInfo.desc.name, appInfo.id, appInfo.retryCount))
+                val execs = idToApp(appId).executors.values
+                if (!execs.exists(_.state == ExecutorState.RUNNING)) {
+                  removeApplication(appInfo, ApplicationState.FAILED)
+                }
+              }
             }
           }
         }

core/src/main/scala/org/apache/spark/deploy/worker/ExecutorRunner.scala

@@ -139,9 +139,10 @@ private[spark] class ExecutorRunner(
       Files.write(header, stderr, Charsets.UTF_8)
       CommandUtils.redirectStream(process.getErrorStream, stderr)
 
-      // Wait for it to exit; this is actually a bad thing if it happens, because we expect to run
-      // long-lived processes only. However, in the future, we might restart the executor a few
-      // times on the same machine.
+      state = ExecutorState.RUNNING
+      worker ! ExecutorStateChanged(appId, execId, state, None, None)
+      // Wait for it to exit; executor may exit with code 0 (when driver instructs it to shutdown)
+      // or with nonzero exit code
       val exitCode = process.waitFor()
       state = ExecutorState.FAILED
       val message = "Command exited with code " + exitCode

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -216,7 +216,7 @@ private[spark] class Worker(
         // ApplicationDescription to be more explicit about this.
         val effectiveSparkHome = Option(execSparkHome_).getOrElse(sparkHome.getAbsolutePath)
         val manager = new ExecutorRunner(appId, execId, appDesc, cores_, memory_,
-          self, workerId, host, new File(effectiveSparkHome), workDir, akkaUrl, ExecutorState.RUNNING)
+          self, workerId, host, new File(effectiveSparkHome), workDir, akkaUrl, ExecutorState.LOADING)
         executors(appId + "/" + execId) = manager
         manager.start()
         coresUsed += cores_