[CARMEL-7539][CARMEL-4980] Cache the RejectedExecutionException if the executor pool is shutdown (apache#189)

Author: wangyum

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -19,8 +19,7 @@ package org.apache.spark.scheduler
 
 import java.io.NotSerializableException
 import java.util.Properties
-import java.util.concurrent.{ConcurrentHashMap, ExecutorService, ScheduledFuture, TimeoutException, TimeUnit}
-import java.util.concurrent.{Future => JFutrue}
+import java.util.concurrent.{ConcurrentHashMap, ExecutorService, Future => JFutrue, RejectedExecutionException, ScheduledFuture, TimeoutException, TimeUnit}
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger, AtomicLong}
 
 import scala.annotation.tailrec
@@ -1908,7 +1907,16 @@ private[spark] class DAGScheduler(
         }
       }
       if (asyncTaskSubmission) {
-        broadcastPool.submit(broadCastTask)
+        try {
+          broadcastPool.submit(broadCastTask)
+        } catch {
+          case _: RejectedExecutionException
+              if broadcastPool.isShutdown && SparkConf.isAnalyticsCluster =>
+            logWarning("Failed to broadcast as executor pool was shutdown")
+            broadCastTask.run()
+          case e: Throwable =>
+            throw e
+        }
       } else {
         broadCastTask.run()
       }

core/src/main/scala/org/apache/spark/scheduler/cluster/AnalyticsCoarseGrainedSchedulerBackend.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.scheduler.cluster
 
-import java.util.concurrent.{CopyOnWriteArrayList, ScheduledExecutorService, TimeUnit}
+import java.util.concurrent.{CopyOnWriteArrayList, RejectedExecutionException, ScheduledExecutorService, TimeUnit}
 import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
 import javax.annotation.concurrent.GuardedBy
 
@@ -413,23 +413,27 @@ class AnalyticsCoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val r
         return
       }
 
-      val f = Future {
-        val startTime = System.currentTimeMillis()
-        val taskDescs = scheduler.resourceOffers(offers)
-        val tasks = taskDescs.flatten
-        if (!tasks.isEmpty) {
-          launchTasks(tasks)
-          logInfo(s"makeOffers launched tasks size: ${tasks.size}," +
-            s" take ${System.currentTimeMillis() - startTime}ms")
-        }
-      }(offerExecutionContext)
-      f.onComplete {
-        case Failure(ex) =>
-          // TODO handle task launch failure, either retry or abort
-          logError(s"makeOffers error with exception: ${ex.getMessage}", ex)
-        case _ =>
-      }(offerExecutionContext)
-
+      try {
+        val f = Future {
+          val startTime = System.currentTimeMillis()
+          val taskDescs = scheduler.resourceOffers(offers)
+          val tasks = taskDescs.flatten
+          if (!tasks.isEmpty) {
+            launchTasks(tasks)
+            logInfo(s"makeOffers launched tasks size: ${tasks.size}," +
+              s" take ${System.currentTimeMillis() - startTime}ms")
+          }
+        }(offerExecutionContext)
+        f.onComplete {
+          case Failure(ex) =>
+            // TODO handle task launch failure, either retry or abort
+            logError(s"makeOffers error with exception: ${ex.getMessage}", ex)
+          case _ =>
+        }(offerExecutionContext)
+      } catch {
+        case _: RejectedExecutionException if offerExecutionContext.isShutdown =>
+          logWarning("Failed to makeOffers as executor pool was shutdown")
+      }
     }
 
     private def buildWorkerOffer(executorId: String, executorData: ExecutorData) = {
@@ -490,22 +494,27 @@ class AnalyticsCoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val r
           return
         }
 
-        val f = Future {
-          val startTime = System.currentTimeMillis()
-          val taskDescs = scheduler.resourceOffers(offers)
-          val tasks = taskDescs.flatten
-          if (tasks.nonEmpty) {
-            launchTasks(tasks)
-            logInfo(s"makeBatchOffers launched tasks size: ${tasks.size}," +
-              s" take ${System.currentTimeMillis() - startTime}ms")
-          }
-        }(offerExecutionContext)
-        f.onComplete {
-          case Failure(ex) =>
-            // TODO handle task launch failure, either retry or abort
-            logError(s"makeBatchOffers error with exception: ${ex.getMessage}", ex)
-          case _ =>
-        }(offerExecutionContext)
+        try {
+          val f = Future {
+            val startTime = System.currentTimeMillis()
+            val taskDescs = scheduler.resourceOffers(offers)
+            val tasks = taskDescs.flatten
+            if (tasks.nonEmpty) {
+              launchTasks(tasks)
+              logInfo(s"makeBatchOffers launched tasks size: ${tasks.size}," +
+                s" take ${System.currentTimeMillis() - startTime}ms")
+            }
+          }(offerExecutionContext)
+          f.onComplete {
+            case Failure(ex) =>
+              // TODO handle task launch failure, either retry or abort
+              logError(s"makeBatchOffers error with exception: ${ex.getMessage}", ex)
+            case _ =>
+          }(offerExecutionContext)
+        } catch {
+          case _: RejectedExecutionException if offerExecutionContext.isShutdown =>
+            logWarning("Failed to makeOffers as executor pool was shutdown")
+        }
       }
     }