Fix dynamic allocation with external shuffle service (apache-spark-on-k8s#445)

Author: rynorris

core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala

@@ -25,7 +25,7 @@ import scala.util.control.{ControlThrowable, NonFatal}
 
 import com.codahale.metrics.{Gauge, MetricRegistry}
 
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{config, Logging}
 import org.apache.spark.internal.config._
 import org.apache.spark.metrics.source.Source
 import org.apache.spark.scheduler._
@@ -118,6 +118,8 @@ private[spark] class ExecutorAllocationManager(
   private val inactiveShuffleExecutorIdleTimeoutS = conf.getTimeAsSeconds(
     "spark.dynamicAllocation.inactiveShuffleExecutorIdleTimeout", s"${Integer.MAX_VALUE}s")
 
+  private val externalShuffleServiceEnabled = conf.get(config.SHUFFLE_SERVICE_ENABLED)
+
   // During testing, the methods to actually kill and add executors are mocked out
   private val testing = conf.getBoolean("spark.dynamicAllocation.testing", false)
 
@@ -599,7 +601,7 @@ private[spark] class ExecutorAllocationManager(
    */
   private def onExecutorIdle(executorId: String): Unit = synchronized {
     if (executorIds.contains(executorId)) {
-      val hasActiveShuffleBlocks =
+      val hasActiveShuffleBlocks = !externalShuffleServiceEnabled &&
         mapOutputTracker.hasOutputsOnExecutor(executorId, activeOnly = true)
       if (!removeTimes.contains(executorId)
         && !executorsPendingToRemove.contains(executorId)
@@ -608,7 +610,8 @@ private[spark] class ExecutorAllocationManager(
         // blocks we are concerned with are reported to the driver. Note that this
         // does not include broadcast blocks.
         val hasCachedBlocks = blockManagerMaster.hasCachedBlocks(executorId)
-        val hasAnyShuffleBlocks = mapOutputTracker.hasOutputsOnExecutor(executorId)
+        val hasAnyShuffleBlocks =
+          !externalShuffleServiceEnabled && mapOutputTracker.hasOutputsOnExecutor(executorId)
         val now = clock.getTimeMillis()
 
         // Use the maximum of all the timeouts that apply.