Simplify ClientBase#monitorApplication

There's only so much abstraction that you could do before it
gets too complicated.

Author: andrewor14

yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ClientBase.scala

@@ -412,17 +412,15 @@ private[spark] trait ClientBase extends Logging {
    *
    * @param returnOnRunning Whether to also return the application state when it is RUNNING.
    * @param logApplicationReport Whether to log details of the application report every iteration.
-   * @param shouldKeepMonitoring The condition to keep monitoring.
    * @return state of the application, one of FINISHED, FAILED, KILLED, and RUNNING.
    */
   def monitorApplication(
       appId: ApplicationId,
       returnOnRunning: Boolean = false,
-      logApplicationReport: Boolean = true,
-      shouldKeepMonitoring: () => Boolean = () => true): YarnApplicationState = {
+      logApplicationReport: Boolean = true): YarnApplicationState = {
     val interval = sparkConf.getLong("spark.yarn.report.interval", 1000)
     var firstIteration = true
-    while (shouldKeepMonitoring()) {
+    while (true) {
       Thread.sleep(interval)
       val report = getApplicationReport(appId)
       val state = report.getYarnApplicationState

yarn/common/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala

@@ -117,25 +117,28 @@ private[spark] class YarnClientSchedulerBackend(
    * If the application has exited for any reason, stop the SparkContext.
    * This assumes both `client` and `appId` have already been set.
    */
-  private def asyncMonitorApplication(): Thread = {
+  private def asyncMonitorApplication(): Unit = {
     assert(client != null && appId != null, "Application has not been submitted yet!")
     val t = new Thread {
       override def run() {
-        val state = client.monitorApplication(
-          appId, logApplicationReport = false, shouldKeepMonitoring = isStopping) // blocking
-        if (state == YarnApplicationState.FINISHED ||
-          state == YarnApplicationState.KILLED ||
-          state == YarnApplicationState.FAILED) {
-          logWarning(s"Yarn application has exited: $state")
-          sc.stop()
-          stopping = true
+        while (!stopping) {
+          val report = client.getApplicationReport(appId)
+          val state = report.getYarnApplicationState()
+          if (state == YarnApplicationState.FINISHED ||
+            state == YarnApplicationState.KILLED ||
+            state == YarnApplicationState.FAILED) {
+            logError(s"Yarn application has already exited with state $state!")
+            sc.stop()
+            stopping = true
+          }
+          Thread.sleep(1000L)
         }
+        Thread.currentThread().interrupt()
       }
     }
-    t.setName("Yarn Application State Monitor")
+    t.setName("Yarn application state monitor")
     t.setDaemon(true)
     t.start()
-    t
   }
 
   /**