revert to the old version and do some robust

Author: WangTaoTheTonic

docs/running-on-yarn.md

@@ -196,6 +196,14 @@ Most of the configs are the same for Spark on YARN as for other deployment modes
   It should be no larger than the global number of max attempts in the YARN configuration.
   </td>
 </tr>
+<tr>
+  <td><code>spark.yarn.waitForCompletion</code></td>
+  <td>true</td>
+  <td>
+  Whether to wait for application to complete in cluster mode. If set this to true, client process will enter a loop to track application's status.
+  Otherwise, client process will exist as long as application is successfully submitted.
+  </td>
+</tr>
 </table>
 
 # Launching Spark on YARN

yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala

@@ -67,7 +67,7 @@ private[spark] class Client(
   private val distCacheMgr = new ClientDistributedCacheManager()
   private val isClusterMode = args.isClusterMode
   private val fireAndForget = isClusterMode &&
-    sparkConf.getBoolean("spark.yarn.cluster.quiet", false)
+    !sparkConf.getBoolean("spark.yarn.submit.waitAppCompletion", true)
 
 
   def stop(): Unit = yarnClient.stop()
@@ -622,13 +622,12 @@ private[spark] class Client(
   def run(): Unit = {
     val appId = submitApplication()
     if (fireAndForget) {
-      logInfo("... waiting before polling ResourceManager for application state")
-      Thread.sleep(5000)
       logInfo("... polling ResourceManager for application state")
-
       val report = getApplicationReport(appId)
       val state = report.getYarnApplicationState
-
+      if (state == YarnApplicationState.FAILED || state == YarnApplicationState.KILLED) {
+        throw new SparkException(s"Application finished with status: $state")
+      }
       logInfo(s"Application report for $appId (state: $state)")
       logInfo(formatReportDetails(report))
     } else {