[CARMEL-7504] Log error message instead of stopping spark context (apache#274)

Author: wakun

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala

@@ -569,17 +569,31 @@ private[spark] class ApplicationMaster(
     var failureCount = 0
     while (!finished) {
       try {
-        if (allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
-          finish(FinalApplicationStatus.FAILED,
-            ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
-            s"Max number of executor failures ($maxNumExecutorFailures) reached")
-        } else if (allocator.isAllNodeExcluded) {
-          finish(FinalApplicationStatus.FAILED,
-            ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
-            "Due to executor failures all available nodes are excluded")
+        if (SparkConf.isETLCluster) {
+          if (allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
+            finish(FinalApplicationStatus.FAILED,
+              ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
+              s"Max number of executor failures ($maxNumExecutorFailures) reached")
+          } else if (allocator.isAllNodeExcluded) {
+            finish(FinalApplicationStatus.FAILED,
+              ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
+              "Due to executor failures all available nodes are excluded")
+          } else {
+            logDebug("Sending progress")
+            allocator.allocateResources()
+          }
         } else {
-          logDebug("Sending progress")
-          allocator.allocateResources()
+          if (allocator.isAllNodeExcluded) {
+            finish(FinalApplicationStatus.FAILED,
+              ApplicationMaster.EXIT_MAX_EXECUTOR_FAILURES,
+              "Due to executor failures all available nodes are excluded")
+          } else {
+            if (allocator.getNumExecutorsFailed >= maxNumExecutorFailures) {
+              logError(s"Max number of executor failures ($maxNumExecutorFailures) reached")
+            }
+            logDebug("Sending progress")
+            allocator.allocateResources()
+          }
         }
         failureCount = 0
       } catch {