This is an automated email from the ASF dual-hosted git repository. irashid pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push: new b312033 [SPARK-20286][CORE] Improve logic for timing out executors in dynamic allocation. b312033 is described below commit b312033bd33cd6cbf1f166ccaa7a5df4e3421078 Author: Marcelo Vanzin <van...@cloudera.com> AuthorDate: Wed Jun 5 08:09:44 2019 -0500 [SPARK-20286][CORE] Improve logic for timing out executors in dynamic allocation. This change refactors the portions of the ExecutorAllocationManager class that track executor state into a new class, to achieve a few goals: - make the code easier to understand - better separate concerns (task backlog vs. executor state) - less synchronization between event and allocation threads - less coupling between the allocation code and executor state tracking The executor tracking code was moved to a new class (ExecutorMonitor) that encapsulates all the logic of tracking what happens to executors and when they can be timed out. The logic to actually remove the executors remains in the EAM, since it still requires information that is not tracked by the new executor monitor code. In the executor monitor itself, of interest, specifically, is a change in how cached blocks are tracked; instead of polling the block manager, the monitor now uses events to track which executors have cached blocks, and is able to detect also unpersist events and adjust the time when the executor should be removed accordingly. (That's the bug mentioned in the PR title.) Because of the refactoring, a few tests in the old EAM test suite were removed, since they're now covered by the newly added test suite. The EAM suite was also changed a little bit to not instantiate a SparkContext every time. This allowed some cleanup, and the tests also run faster. Tested with new and updated unit tests, and with multiple TPC-DS workloads running with dynamic allocation on; also some manual tests for the caching behavior. Closes #24704 from vanzin/SPARK-20286. Authored-by: Marcelo Vanzin <van...@cloudera.com> Signed-off-by: Imran Rashid <iras...@cloudera.com> --- .../apache/spark/ExecutorAllocationClient.scala | 9 +- .../apache/spark/ExecutorAllocationManager.scala | 239 +------ .../main/scala/org/apache/spark/SparkContext.scala | 3 +- .../org/apache/spark/internal/config/package.scala | 8 +- .../cluster/CoarseGrainedSchedulerBackend.scala | 4 + .../spark/scheduler/dynalloc/ExecutorMonitor.scala | 283 ++++++++ .../apache/spark/storage/BlockManagerMaster.scala | 9 - .../spark/storage/BlockManagerMasterEndpoint.scala | 31 - .../spark/storage/BlockManagerMessages.scala | 3 - ...g.apache.spark.scheduler.ExternalClusterManager | 1 - .../spark/ExecutorAllocationManagerSuite.scala | 708 ++++++--------------- .../deploy/ExternalShuffleServiceDbSuite.scala | 9 +- .../scheduler/dynalloc/ExecutorMonitorSuite.scala | 289 +++++++++ .../spark/storage/BlockManagerInfoSuite.scala | 14 - .../scheduler/ExecutorAllocationManagerSuite.scala | 2 +- 15 files changed, 822 insertions(+), 790 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala index 63d87b4..cb965cb 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala @@ -23,11 +23,18 @@ package org.apache.spark */ private[spark] trait ExecutorAllocationClient { - /** Get the list of currently active executors */ private[spark] def getExecutorIds(): Seq[String] /** + * Whether an executor is active. An executor is active when it can be used to execute tasks + * for jobs submitted by the application. + * + * @return whether the executor with the given ID is currently active. + */ + def isExecutorActive(id: String): Boolean + + /** * Update the cluster manager on our scheduling needs. Three bits of information are included * to help it make decisions. * @param numExecutors The total number of executors we'd like to have. The cluster manager diff --git a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala index 1782027..63df7cc 100644 --- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala +++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala @@ -30,7 +30,7 @@ import org.apache.spark.internal.config._ import org.apache.spark.internal.config.Tests.TEST_SCHEDULE_INTERVAL import org.apache.spark.metrics.source.Source import org.apache.spark.scheduler._ -import org.apache.spark.storage.BlockManagerMaster +import org.apache.spark.scheduler.dynalloc.ExecutorMonitor import org.apache.spark.util.{Clock, SystemClock, ThreadUtils, Utils} /** @@ -94,7 +94,7 @@ private[spark] class ExecutorAllocationManager( client: ExecutorAllocationClient, listenerBus: LiveListenerBus, conf: SparkConf, - blockManagerMaster: BlockManagerMaster) + clock: Clock = new SystemClock()) extends Logging { allocationManager => @@ -113,11 +113,6 @@ private[spark] class ExecutorAllocationManager( private val sustainedSchedulerBacklogTimeoutS = conf.get(DYN_ALLOCATION_SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT) - // How long an executor must be idle for before it is removed (seconds) - private val executorIdleTimeoutS = conf.get(DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT) - - private val cachedExecutorIdleTimeoutS = conf.get(DYN_ALLOCATION_CACHED_EXECUTOR_IDLE_TIMEOUT) - // During testing, the methods to actually kill and add executors are mocked out private val testing = conf.get(DYN_ALLOCATION_TESTING) @@ -139,20 +134,10 @@ private[spark] class ExecutorAllocationManager( // is the number of executors we would immediately want from the cluster manager. private var numExecutorsTarget = initialNumExecutors - // Executors that have been requested to be removed but have not been killed yet - private val executorsPendingToRemove = new mutable.HashSet[String] - - // All known executors - private val executorIds = new mutable.HashSet[String] - // A timestamp of when an addition should be triggered, or NOT_SET if it is not set // This is set when pending tasks are added but not scheduled yet private var addTime: Long = NOT_SET - // A timestamp for each executor of when the executor should be removed, indexed by the ID - // This is set when an executor is no longer running a task, or when it first registers - private val removeTimes = new mutable.HashMap[String, Long] - // Polling loop interval (ms) private val intervalMillis: Long = if (Utils.isTesting) { conf.get(TEST_SCHEDULE_INTERVAL) @@ -160,12 +145,11 @@ private[spark] class ExecutorAllocationManager( 100 } - // Clock used to schedule when executors should be added and removed - private var clock: Clock = new SystemClock() - // Listener for Spark events that impact the allocation policy val listener = new ExecutorAllocationListener + val executorMonitor = new ExecutorMonitor(conf, client, clock) + // Executor that handles the scheduling task. private val executor = ThreadUtils.newDaemonSingleThreadScheduledExecutor("spark-dynamic-executor-allocation") @@ -210,12 +194,6 @@ private[spark] class ExecutorAllocationManager( throw new SparkException( s"s${DYN_ALLOCATION_SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT.key} must be > 0!") } - if (executorIdleTimeoutS < 0) { - throw new SparkException(s"${DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT.key} must be >= 0!") - } - if (cachedExecutorIdleTimeoutS < 0) { - throw new SparkException(s"${DYN_ALLOCATION_CACHED_EXECUTOR_IDLE_TIMEOUT.key} must be >= 0!") - } // Require external shuffle service for dynamic allocation // Otherwise, we may lose shuffle files when killing executors if (!conf.get(config.SHUFFLE_SERVICE_ENABLED) && !testing) { @@ -230,18 +208,12 @@ private[spark] class ExecutorAllocationManager( } /** - * Use a different clock for this allocation manager. This is mainly used for testing. - */ - def setClock(newClock: Clock): Unit = { - clock = newClock - } - - /** * Register for scheduler callbacks to decide when to add and remove executors, and start * the scheduling task. */ def start(): Unit = { listenerBus.addToManagementQueue(listener) + listenerBus.addToManagementQueue(executorMonitor) val scheduleTask = new Runnable() { override def run(): Unit = { @@ -278,8 +250,7 @@ private[spark] class ExecutorAllocationManager( def reset(): Unit = synchronized { addTime = 0L numExecutorsTarget = initialNumExecutors - executorsPendingToRemove.clear() - removeTimes.clear() + executorMonitor.reset() } /** @@ -307,19 +278,13 @@ private[spark] class ExecutorAllocationManager( * This is factored out into its own method for testing. */ private def schedule(): Unit = synchronized { - val now = clock.getTimeMillis - - val executorIdsToBeRemoved = ArrayBuffer[String]() - removeTimes.retain { case (executorId, expireTime) => - val expired = now >= expireTime - if (expired) { - initializing = false - executorIdsToBeRemoved += executorId - } - !expired + val executorIdsToBeRemoved = executorMonitor.timedOutExecutors() + if (executorIdsToBeRemoved.nonEmpty) { + initializing = false } + // Update executor target number only after initializing flag is unset - updateAndSyncNumExecutorsTarget(now) + updateAndSyncNumExecutorsTarget(clock.getTimeMillis()) if (executorIdsToBeRemoved.nonEmpty) { removeExecutors(executorIdsToBeRemoved) } @@ -395,7 +360,7 @@ private[spark] class ExecutorAllocationManager( val oldNumExecutorsTarget = numExecutorsTarget // There's no point in wasting time ramping up to the number of executors we already have, so // make sure our target is at least as much as our current allocation: - numExecutorsTarget = math.max(numExecutorsTarget, executorIds.size) + numExecutorsTarget = math.max(numExecutorsTarget, executorMonitor.executorCount) // Boost our target with the number to add for this round: numExecutorsTarget += numExecutorsToAdd // Ensure that our target doesn't exceed what we need at the present moment: @@ -455,7 +420,7 @@ private[spark] class ExecutorAllocationManager( val executorIdsToBeRemoved = new ArrayBuffer[String] logInfo("Request to remove executorIds: " + executors.mkString(", ")) - val numExistingExecutors = allocationManager.executorIds.size - executorsPendingToRemove.size + val numExistingExecutors = executorMonitor.executorCount - executorMonitor.pendingRemovalCount var newExecutorTotal = numExistingExecutors executors.foreach { executorIdToBeRemoved => @@ -465,7 +430,7 @@ private[spark] class ExecutorAllocationManager( } else if (newExecutorTotal - 1 < numExecutorsTarget) { logDebug(s"Not removing idle executor $executorIdToBeRemoved because there are only " + s"$newExecutorTotal executor(s) left (number of executor target $numExecutorsTarget)") - } else if (canBeKilled(executorIdToBeRemoved)) { + } else { executorIdsToBeRemoved += executorIdToBeRemoved newExecutorTotal -= 1 } @@ -484,24 +449,17 @@ private[spark] class ExecutorAllocationManager( client.killExecutors(executorIdsToBeRemoved, adjustTargetNumExecutors = false, countFailures = false, force = false) } + // [SPARK-21834] killExecutors api reduces the target number of executors. // So we need to update the target with desired value. client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks, hostToLocalTaskCount) // reset the newExecutorTotal to the existing number of executors newExecutorTotal = numExistingExecutors if (testing || executorsRemoved.nonEmpty) { - executorsRemoved.foreach { removedExecutorId => - // If it has an exclusive cached block then cachedExecutorIdleTimeoutS is used for timeout - val idleTimeout = if (blockManagerMaster.hasExclusiveCachedBlocks(removedExecutorId)) { - cachedExecutorIdleTimeoutS - } else { - executorIdleTimeoutS - } - newExecutorTotal -= 1 - logInfo(s"Removing executor $removedExecutorId because it has been idle for " + - s"$idleTimeout seconds (new desired total will be $newExecutorTotal)") - executorsPendingToRemove.add(removedExecutorId) - } + newExecutorTotal -= executorsRemoved.size + executorMonitor.executorsKilled(executorsRemoved) + logInfo(s"Executors ${executorsRemoved.mkString(",")} removed due to idle timeout." + + s"(new desired total will be $newExecutorTotal)") executorsRemoved } else { logWarning(s"Unable to reach the cluster manager to kill executor/s " + @@ -511,70 +469,6 @@ private[spark] class ExecutorAllocationManager( } /** - * Request the cluster manager to remove the given executor. - * Return whether the request is acknowledged. - */ - private def removeExecutor(executorId: String): Boolean = synchronized { - val executorsRemoved = removeExecutors(Seq(executorId)) - executorsRemoved.nonEmpty && executorsRemoved(0) == executorId - } - - /** - * Determine if the given executor can be killed. - */ - private def canBeKilled(executorId: String): Boolean = synchronized { - // Do not kill the executor if we are not aware of it (should never happen) - if (!executorIds.contains(executorId)) { - logWarning(s"Attempted to remove unknown executor $executorId!") - return false - } - - // Do not kill the executor again if it is already pending to be killed (should never happen) - if (executorsPendingToRemove.contains(executorId)) { - logWarning(s"Attempted to remove executor $executorId " + - s"when it is already pending to be removed!") - return false - } - - true - } - - /** - * Callback invoked when the specified executor has been added. - */ - private def onExecutorAdded(executorId: String): Unit = synchronized { - if (!executorIds.contains(executorId)) { - executorIds.add(executorId) - // If an executor (call this executor X) is not removed because the lower bound - // has been reached, it will no longer be marked as idle. When new executors join, - // however, we are no longer at the lower bound, and so we must mark executor X - // as idle again so as not to forget that it is a candidate for removal. (see SPARK-4951) - executorIds.filter(listener.isExecutorIdle).foreach(onExecutorIdle) - logInfo(s"New executor $executorId has registered (new total is ${executorIds.size})") - } else { - logWarning(s"Duplicate executor $executorId has registered") - } - } - - /** - * Callback invoked when the specified executor has been removed. - */ - private def onExecutorRemoved(executorId: String): Unit = synchronized { - if (executorIds.contains(executorId)) { - executorIds.remove(executorId) - removeTimes.remove(executorId) - logInfo(s"Existing executor $executorId has been removed (new total is ${executorIds.size})") - if (executorsPendingToRemove.contains(executorId)) { - executorsPendingToRemove.remove(executorId) - logDebug(s"Executor $executorId is no longer pending to " + - s"be removed (${executorsPendingToRemove.size} left)") - } - } else { - logWarning(s"Unknown executor $executorId has been removed!") - } - } - - /** * Callback invoked when the scheduler receives new pending tasks. * This sets a time in the future that decides when executors should be added * if it is not already set. @@ -598,46 +492,6 @@ private[spark] class ExecutorAllocationManager( } /** - * Callback invoked when the specified executor is no longer running any tasks. - * This sets a time in the future that decides when this executor should be removed if - * the executor is not already marked as idle. - */ - private def onExecutorIdle(executorId: String): Unit = synchronized { - if (executorIds.contains(executorId)) { - if (!removeTimes.contains(executorId) && !executorsPendingToRemove.contains(executorId)) { - // Note that it is not necessary to query the executors since all the cached blocks we are - // concerned with are reported to the driver. This does not include broadcast blocks and - // non-exclusive blocks which are also available via the external shuffle service. - val hasCachedBlocks = blockManagerMaster.hasExclusiveCachedBlocks(executorId) - val now = clock.getTimeMillis() - val timeout = { - if (hasCachedBlocks) { - // Use a different timeout if the executor has cached blocks. - now + cachedExecutorIdleTimeoutS * 1000 - } else { - now + executorIdleTimeoutS * 1000 - } - } - val realTimeout = if (timeout <= 0) Long.MaxValue else timeout // overflow - removeTimes(executorId) = realTimeout - logDebug(s"Starting idle timer for $executorId because there are no more tasks " + - s"scheduled to run on the executor (to expire in ${(realTimeout - now)/1000} seconds)") - } - } else { - logWarning(s"Attempted to mark unknown executor $executorId idle") - } - } - - /** - * Callback invoked when the specified executor is now running a task. - * This resets all variables used for removing this executor. - */ - private def onExecutorBusy(executorId: String): Unit = synchronized { - logDebug(s"Clearing idle timer for $executorId because it is now running a task") - removeTimes.remove(executorId) - } - - /** * A listener that notifies the given allocation manager of when to add and remove executors. * * This class is intentionally conservative in its assumptions about the relative ordering @@ -650,7 +504,6 @@ private[spark] class ExecutorAllocationManager( // Should be 0 when no stages are active. private val stageIdToNumRunningTask = new mutable.HashMap[Int, Int] private val stageIdToTaskIndices = new mutable.HashMap[Int, mutable.HashSet[Int]] - private val executorIdToTaskIds = new mutable.HashMap[String, mutable.HashSet[Long]] // Number of speculative tasks to be scheduled in each stage private val stageIdToNumSpeculativeTasks = new mutable.HashMap[Int, Int] // The speculative tasks started in each stage @@ -716,23 +569,11 @@ private[spark] class ExecutorAllocationManager( val stageId = taskStart.stageId val taskId = taskStart.taskInfo.taskId val taskIndex = taskStart.taskInfo.index - val executorId = taskStart.taskInfo.executorId allocationManager.synchronized { if (stageIdToNumRunningTask.contains(stageId)) { stageIdToNumRunningTask(stageId) += 1 } - // This guards against the following race condition: - // 1. The `SparkListenerTaskStart` event is posted before the - // `SparkListenerExecutorAdded` event - // 2. The `SparkListenerExecutorRemoved` event is posted before the - // `SparkListenerTaskStart` event - // Above cases are possible because these events are posted in different threads. - // (see SPARK-4951 SPARK-26927) - if (!allocationManager.executorIds.contains(executorId) && - client.getExecutorIds().contains(executorId)) { - allocationManager.onExecutorAdded(executorId) - } // If this is the last pending task, mark the scheduler queue as empty if (taskStart.taskInfo.speculative) { @@ -744,15 +585,10 @@ private[spark] class ExecutorAllocationManager( if (totalPendingTasks() == 0) { allocationManager.onSchedulerQueueEmpty() } - - // Mark the executor on which this task is scheduled as busy - executorIdToTaskIds.getOrElseUpdate(executorId, new mutable.HashSet[Long]) += taskId - allocationManager.onExecutorBusy(executorId) } } override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = { - val executorId = taskEnd.taskInfo.executorId val taskId = taskEnd.taskInfo.taskId val taskIndex = taskEnd.taskInfo.index val stageId = taskEnd.stageId @@ -760,14 +596,6 @@ private[spark] class ExecutorAllocationManager( if (stageIdToNumRunningTask.contains(stageId)) { stageIdToNumRunningTask(stageId) -= 1 } - // If the executor is no longer running any scheduled tasks, mark it as idle - if (executorIdToTaskIds.contains(executorId)) { - executorIdToTaskIds(executorId) -= taskId - if (executorIdToTaskIds(executorId).isEmpty) { - executorIdToTaskIds -= executorId - allocationManager.onExecutorIdle(executorId) - } - } // If the task failed, we expect it to be resubmitted later. To ensure we have // enough resources to run the resubmitted task, we need to mark the scheduler @@ -785,22 +613,6 @@ private[spark] class ExecutorAllocationManager( } } - override def onExecutorAdded(executorAdded: SparkListenerExecutorAdded): Unit = { - val executorId = executorAdded.executorId - if (executorId != SparkContext.DRIVER_IDENTIFIER) { - // This guards against the race condition in which the `SparkListenerTaskStart` - // event is posted before the `SparkListenerBlockManagerAdded` event, which is - // possible because these events are posted in different threads. (see SPARK-4951) - if (!allocationManager.executorIds.contains(executorId)) { - allocationManager.onExecutorAdded(executorId) - } - } - } - - override def onExecutorRemoved(executorRemoved: SparkListenerExecutorRemoved): Unit = { - allocationManager.onExecutorRemoved(executorRemoved.executorId) - } - override def onSpeculativeTaskSubmitted(speculativeTask: SparkListenerSpeculativeTaskSubmitted) : Unit = { val stageId = speculativeTask.stageId @@ -842,15 +654,6 @@ private[spark] class ExecutorAllocationManager( } /** - * Return true if an executor is not currently running a task, and false otherwise. - * - * Note: This is not thread-safe without the caller owning the `allocationManager` lock. - */ - def isExecutorIdle(executorId: String): Boolean = { - !executorIdToTaskIds.contains(executorId) - } - - /** * Update the Executor placement hints (the number of tasks with locality preferences, * a map where each pair is a node and the number of tasks that would like to be scheduled * on that node). @@ -892,8 +695,8 @@ private[spark] class ExecutorAllocationManager( } registerGauge("numberExecutorsToAdd", numExecutorsToAdd, 0) - registerGauge("numberExecutorsPendingToRemove", executorsPendingToRemove.size, 0) - registerGauge("numberAllExecutors", executorIds.size, 0) + registerGauge("numberExecutorsPendingToRemove", executorMonitor.pendingRemovalCount, 0) + registerGauge("numberAllExecutors", executorMonitor.executorCount, 0) registerGauge("numberTargetExecutors", numExecutorsTarget, 0) registerGauge("numberMaxNeededExecutors", maxNumExecutorsNeeded(), 0) } diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala index b00bb9a..66f8f41 100644 --- a/core/src/main/scala/org/apache/spark/SparkContext.scala +++ b/core/src/main/scala/org/apache/spark/SparkContext.scala @@ -585,8 +585,7 @@ class SparkContext(config: SparkConf) extends Logging { schedulerBackend match { case b: ExecutorAllocationClient => Some(new ExecutorAllocationManager( - schedulerBackend.asInstanceOf[ExecutorAllocationClient], listenerBus, _conf, - _env.blockManager.master)) + schedulerBackend.asInstanceOf[ExecutorAllocationClient], listenerBus, _conf)) case _ => None } diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 32221ee..90826bb 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -357,11 +357,15 @@ package object config { private[spark] val DYN_ALLOCATION_CACHED_EXECUTOR_IDLE_TIMEOUT = ConfigBuilder("spark.dynamicAllocation.cachedExecutorIdleTimeout") - .timeConf(TimeUnit.SECONDS).createWithDefault(Integer.MAX_VALUE) + .timeConf(TimeUnit.SECONDS) + .checkValue(_ >= 0L, "Timeout must be >= 0.") + .createWithDefault(Integer.MAX_VALUE) private[spark] val DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT = ConfigBuilder("spark.dynamicAllocation.executorIdleTimeout") - .timeConf(TimeUnit.SECONDS).createWithDefault(60) + .timeConf(TimeUnit.SECONDS) + .checkValue(_ >= 0L, "Timeout must be >= 0.") + .createWithDefault(60) private[spark] val DYN_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT = ConfigBuilder("spark.dynamicAllocation.schedulerBacklogTimeout") diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index 9f53588..5f8c277 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -540,6 +540,10 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp executorDataMap.keySet.toSeq } + override def isExecutorActive(id: String): Boolean = synchronized { + executorDataMap.contains(id) && !executorsPendingToRemove.contains(id) + } + override def maxNumConcurrentTasks(): Int = { executorDataMap.values.map { executor => executor.totalCores / scheduler.CPUS_PER_TASK diff --git a/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala b/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala new file mode 100644 index 0000000..9aac4d2 --- /dev/null +++ b/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala @@ -0,0 +1,283 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.scheduler.dynalloc + +import java.util.concurrent.{ConcurrentHashMap, TimeUnit} +import java.util.concurrent.atomic.AtomicLong + +import scala.collection.JavaConverters._ +import scala.collection.mutable + +import org.apache.spark._ +import org.apache.spark.internal.Logging +import org.apache.spark.internal.config._ +import org.apache.spark.scheduler._ +import org.apache.spark.storage.RDDBlockId +import org.apache.spark.util.Clock + +/** + * A monitor for executor activity, used by ExecutorAllocationManager to detect idle executors. + */ +private[spark] class ExecutorMonitor( + conf: SparkConf, + client: ExecutorAllocationClient, + clock: Clock) extends SparkListener with Logging { + + private val idleTimeoutMs = TimeUnit.SECONDS.toMillis( + conf.get(DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT)) + private val storageTimeoutMs = TimeUnit.SECONDS.toMillis( + conf.get(DYN_ALLOCATION_CACHED_EXECUTOR_IDLE_TIMEOUT)) + private val fetchFromShuffleSvcEnabled = conf.get(SHUFFLE_SERVICE_ENABLED) && + conf.get(SHUFFLE_SERVICE_FETCH_RDD_ENABLED) + + private val executors = new ConcurrentHashMap[String, Tracker]() + + // The following fields are an optimization to avoid having to scan all executors on every EAM + // schedule interval to find out which ones are timed out. They keep track of when the next + // executor timeout is expected to happen, and the current list of timed out executors. There's + // also a flag that forces the EAM task to recompute the timed out executors, in case some event + // arrives on the listener bus that may cause the current list of timed out executors to change. + // + // There's also per-executor state used for this purpose, so that recomputations can be triggered + // only when really necessary. + // + // Note that this isn't meant to, and cannot, always make the right decision about which executors + // are indeed timed out. For example, the EAM thread may detect a timed out executor while a new + // "task start" event has just been posted to the listener bus and hasn't yet been delivered to + // this listener. There are safeguards in other parts of the code that would prevent that executor + // from being removed. + private val nextTimeout = new AtomicLong(Long.MaxValue) + private var timedOutExecs = Seq.empty[String] + + def reset(): Unit = { + executors.clear() + nextTimeout.set(Long.MaxValue) + timedOutExecs = Nil + } + + /** + * Returns the list of executors that are currently considered to be timed out. + * Should only be called from the EAM thread. + */ + def timedOutExecutors(): Seq[String] = { + val now = clock.getTimeMillis() + if (now >= nextTimeout.get()) { + // Temporarily set the next timeout at Long.MaxValue. This ensures that after + // scanning all executors below, we know when the next timeout for non-timed out + // executors is (whether that update came from the scan, or from a new event + // arriving in a different thread). + nextTimeout.set(Long.MaxValue) + + var newNextTimeout = Long.MaxValue + timedOutExecs = executors.asScala + .filter { case (_, exec) => !exec.pendingRemoval } + .filter { case (_, exec) => + val deadline = exec.timeoutAt + if (deadline > now) { + newNextTimeout = math.min(newNextTimeout, deadline) + exec.timedOut = false + false + } else { + exec.timedOut = true + true + } + } + .keys + .toSeq + updateNextTimeout(newNextTimeout) + } + timedOutExecs + } + + /** + * Mark the given executors as pending to be removed. Should only be called in the EAM thread. + */ + def executorsKilled(ids: Seq[String]): Unit = { + ids.foreach { id => + val tracker = executors.get(id) + if (tracker != null) { + tracker.pendingRemoval = true + } + } + + // Recompute timed out executors in the next EAM callback, since this call invalidates + // the current list. + nextTimeout.set(Long.MinValue) + } + + def executorCount: Int = executors.size() + + def pendingRemovalCount: Int = executors.asScala.count { case (_, exec) => exec.pendingRemoval } + + override def onTaskStart(event: SparkListenerTaskStart): Unit = { + val executorId = event.taskInfo.executorId + // Guard against a late arriving task start event (SPARK-26927). + if (client.isExecutorActive(executorId)) { + val exec = ensureExecutorIsTracked(executorId) + exec.updateRunningTasks(1) + } + } + + override def onTaskEnd(event: SparkListenerTaskEnd): Unit = { + val executorId = event.taskInfo.executorId + val exec = executors.get(executorId) + if (exec != null) { + exec.updateRunningTasks(-1) + } + } + + override def onExecutorAdded(event: SparkListenerExecutorAdded): Unit = { + val exec = ensureExecutorIsTracked(event.executorId) + exec.updateRunningTasks(0) + logInfo(s"New executor ${event.executorId} has registered (new total is ${executors.size()})") + } + + override def onExecutorRemoved(event: SparkListenerExecutorRemoved): Unit = { + val removed = executors.remove(event.executorId) + if (removed != null) { + logInfo(s"Executor ${event.executorId} removed (new total is ${executors.size()})") + if (!removed.pendingRemoval) { + nextTimeout.set(Long.MinValue) + } + } + } + + override def onBlockUpdated(event: SparkListenerBlockUpdated): Unit = { + if (!event.blockUpdatedInfo.blockId.isInstanceOf[RDDBlockId]) { + return + } + + val exec = ensureExecutorIsTracked(event.blockUpdatedInfo.blockManagerId.executorId) + val storageLevel = event.blockUpdatedInfo.storageLevel + val blockId = event.blockUpdatedInfo.blockId.asInstanceOf[RDDBlockId] + + // SPARK-27677. When a block can be fetched from the external shuffle service, the executor can + // be removed without hurting the application too much, since the cached data is still + // available. So don't count blocks that can be served by the external service. + if (storageLevel.isValid && (!fetchFromShuffleSvcEnabled || !storageLevel.useDisk)) { + val hadCachedBlocks = exec.cachedBlocks.nonEmpty + + val blocks = exec.cachedBlocks.getOrElseUpdate(blockId.rddId, + new mutable.BitSet(blockId.splitIndex)) + blocks += blockId.splitIndex + + if (!hadCachedBlocks) { + exec.updateTimeout() + } + } else { + exec.cachedBlocks.get(blockId.rddId).foreach { blocks => + blocks -= blockId.splitIndex + if (blocks.isEmpty) { + exec.cachedBlocks -= blockId.rddId + if (exec.cachedBlocks.isEmpty) { + exec.updateTimeout() + } + } + } + } + } + + override def onUnpersistRDD(event: SparkListenerUnpersistRDD): Unit = { + executors.values().asScala.foreach { exec => + exec.cachedBlocks -= event.rddId + if (exec.cachedBlocks.isEmpty) { + exec.updateTimeout() + } + } + } + + // Visible for testing. + private[dynalloc] def isExecutorIdle(id: String): Boolean = { + Option(executors.get(id)).map(_.isIdle).getOrElse(throw new NoSuchElementException(id)) + } + + // Visible for testing + private[dynalloc] def timedOutExecutors(when: Long): Seq[String] = { + executors.asScala.flatMap { case (id, tracker) => + if (tracker.timeoutAt <= when) Some(id) else None + }.toSeq + } + + // Visible for testing + def executorsPendingToRemove(): Set[String] = { + executors.asScala.filter { case (_, exec) => exec.pendingRemoval }.keys.toSet + } + + /** + * This method should be used when updating executor state. It guards against a race condition in + * which the `SparkListenerTaskStart` event is posted before the `SparkListenerBlockManagerAdded` + * event, which is possible because these events are posted in different threads. (see SPARK-4951) + */ + private def ensureExecutorIsTracked(id: String): Tracker = { + executors.computeIfAbsent(id, _ => new Tracker()) + } + + private def updateNextTimeout(newValue: Long): Unit = { + while (true) { + val current = nextTimeout.get() + if (newValue >= current || nextTimeout.compareAndSet(current, newValue)) { + return + } + } + } + + private class Tracker { + @volatile var timeoutAt: Long = Long.MaxValue + + // Tracks whether this executor is thought to be timed out. It's used to detect when the list + // of timed out executors needs to be updated due to the executor's state changing. + @volatile var timedOut: Boolean = false + + var pendingRemoval: Boolean = false + + private var idleStart: Long = -1 + private var runningTasks: Int = 0 + + // Maps RDD IDs to the partition IDs stored in the executor. + // This should only be used in the event thread. + val cachedBlocks = new mutable.HashMap[Int, mutable.BitSet]() + + // For testing. + def isIdle: Boolean = idleStart >= 0 + + def updateRunningTasks(delta: Int): Unit = { + runningTasks = math.max(0, runningTasks + delta) + idleStart = if (runningTasks == 0) clock.getTimeMillis() else -1L + updateTimeout() + } + + def updateTimeout(): Unit = { + val oldDeadline = timeoutAt + val newDeadline = if (idleStart >= 0) { + idleStart + (if (cachedBlocks.nonEmpty) storageTimeoutMs else idleTimeoutMs) + } else { + Long.MaxValue + } + + timeoutAt = newDeadline + + // If the executor was thought to be timed out, but the new deadline is later than the + // old one, ask the EAM thread to update the list of timed out executors. + if (newDeadline > oldDeadline && timedOut) { + nextTimeout.set(Long.MinValue) + } else { + updateNextTimeout(newDeadline) + } + } + } +} diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala index b18d38f..939a134 100644 --- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala +++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala @@ -222,15 +222,6 @@ class BlockManagerMaster( timeout.awaitResult(future) } - /** - * Find out if the executor has cached blocks which are not available via the external shuffle - * service. - * This method does not consider broadcast blocks, since they are not reported to the master. - */ - def hasExclusiveCachedBlocks(executorId: String): Boolean = { - driverEndpoint.askSync[Boolean](HasExclusiveCachedBlocks(executorId)) - } - /** Stop the driver endpoint, called only on the Spark driver node */ def stop() { if (driverEndpoint != null && isDriver) { diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala index 65ec1c3..2057d1b 100644 --- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala +++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala @@ -147,18 +147,6 @@ class BlockManagerMasterEndpoint( case BlockManagerHeartbeat(blockManagerId) => context.reply(heartbeatReceived(blockManagerId)) - - case HasExclusiveCachedBlocks(executorId) => - blockManagerIdByExecutor.get(executorId) match { - case Some(bm) => - if (blockManagerInfo.contains(bm)) { - val bmInfo = blockManagerInfo(bm) - context.reply(bmInfo.exclusiveCachedBlocks.nonEmpty) - } else { - context.reply(false) - } - case None => context.reply(false) - } } private def removeRdd(rddId: Int): Future[Seq[Int]] = { @@ -589,12 +577,6 @@ private[spark] class BlockManagerInfo( // Mapping from block id to its status. private val _blocks = new JHashMap[BlockId, BlockStatus] - /** - * Cached blocks which are not available via the external shuffle service. - * This does not include broadcast blocks. - */ - private val _exclusiveCachedBlocks = new mutable.HashSet[BlockId] - def getStatus(blockId: BlockId): Option[BlockStatus] = Option(_blocks.get(blockId)) def updateLastSeenMs() { @@ -662,15 +644,6 @@ private[spark] class BlockManagerInfo( } } - if (!blockId.isBroadcast) { - if (!externalShuffleServiceEnabled || !storageLevel.useDisk) { - _exclusiveCachedBlocks += blockId - } else if (blockExists) { - // removing block from the exclusive cached blocks when updated to non-exclusive - _exclusiveCachedBlocks -= blockId - } - } - externalShuffleServiceBlockStatus.foreach { shuffleServiceBlocks => if (!blockId.isBroadcast && blockStatus.diskSize > 0) { shuffleServiceBlocks.put(blockId, blockStatus) @@ -679,7 +652,6 @@ private[spark] class BlockManagerInfo( } else if (blockExists) { // If isValid is not true, drop the block. _blocks.remove(blockId) - _exclusiveCachedBlocks -= blockId externalShuffleServiceBlockStatus.foreach { blockStatus => blockStatus.remove(blockId) } @@ -703,7 +675,6 @@ private[spark] class BlockManagerInfo( blockStatus.remove(blockId) } } - _exclusiveCachedBlocks -= blockId } def remainingMem: Long = _remainingMem @@ -712,8 +683,6 @@ private[spark] class BlockManagerInfo( def blocks: JHashMap[BlockId, BlockStatus] = _blocks - def exclusiveCachedBlocks: collection.Set[BlockId] = _exclusiveCachedBlocks - override def toString: String = "BlockManagerInfo " + timeMs + " " + _remainingMem def clear() { diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala index 3dbac69..382501a 100644 --- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala +++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala @@ -122,8 +122,5 @@ private[spark] object BlockManagerMessages { case class BlockManagerHeartbeat(blockManagerId: BlockManagerId) extends ToBlockManagerMaster - case class HasExclusiveCachedBlocks(executorId: String) extends ToBlockManagerMaster - case class IsExecutorAlive(executorId: String) extends ToBlockManagerMaster - } diff --git a/core/src/test/resources/META-INF/services/org.apache.spark.scheduler.ExternalClusterManager b/core/src/test/resources/META-INF/services/org.apache.spark.scheduler.ExternalClusterManager index 1c78f1a..60054c8 100644 --- a/core/src/test/resources/META-INF/services/org.apache.spark.scheduler.ExternalClusterManager +++ b/core/src/test/resources/META-INF/services/org.apache.spark.scheduler.ExternalClusterManager @@ -1,4 +1,3 @@ org.apache.spark.scheduler.DummyExternalClusterManager org.apache.spark.scheduler.MockExternalClusterManager -org.apache.spark.DummyLocalExternalClusterManager org.apache.spark.scheduler.CSMockExternalClusterManager diff --git a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala index 9c30124..2b75f2e 100644 --- a/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala @@ -23,93 +23,95 @@ import org.mockito.ArgumentMatchers.{any, eq => meq} import org.mockito.Mockito.{mock, never, verify, when} import org.scalatest.PrivateMethodTester -import org.apache.spark.executor.TaskMetrics import org.apache.spark.internal.config import org.apache.spark.internal.config.Tests.TEST_SCHEDULE_INTERVAL +import org.apache.spark.metrics.MetricsSystem import org.apache.spark.scheduler._ -import org.apache.spark.scheduler.ExternalClusterManager import org.apache.spark.scheduler.cluster.ExecutorInfo -import org.apache.spark.scheduler.local.LocalSchedulerBackend -import org.apache.spark.storage.BlockManagerMaster -import org.apache.spark.util.ManualClock +import org.apache.spark.util.{Clock, ManualClock, SystemClock} /** * Test add and remove behavior of ExecutorAllocationManager. */ -class ExecutorAllocationManagerSuite - extends SparkFunSuite - with LocalSparkContext { +class ExecutorAllocationManagerSuite extends SparkFunSuite { import ExecutorAllocationManager._ import ExecutorAllocationManagerSuite._ - private val contexts = new mutable.ListBuffer[SparkContext]() + private val managers = new mutable.ListBuffer[ExecutorAllocationManager]() + private var listenerBus: LiveListenerBus = _ + private var client: ExecutorAllocationClient = _ override def beforeEach(): Unit = { super.beforeEach() - contexts.clear() + managers.clear() + listenerBus = new LiveListenerBus(new SparkConf()) + listenerBus.start(null, mock(classOf[MetricsSystem])) + client = mock(classOf[ExecutorAllocationClient]) + when(client.isExecutorActive(any())).thenReturn(true) } override def afterEach(): Unit = { try { - contexts.foreach(_.stop()) + listenerBus.stop() + managers.foreach(_.stop()) } finally { + listenerBus = null super.afterEach() } } - private def post(bus: LiveListenerBus, event: SparkListenerEvent): Unit = { - bus.post(event) - bus.waitUntilEmpty(1000) + private def post(event: SparkListenerEvent): Unit = { + listenerBus.post(event) + listenerBus.waitUntilEmpty(1000) } - test("verify min/max executors") { - val conf = new SparkConf() - .setMaster("myDummyLocalExternalClusterManager") - .setAppName("test-executor-allocation-manager") - .set(config.DYN_ALLOCATION_ENABLED, true) - .set(config.DYN_ALLOCATION_TESTING, true) + test("initialize dynamic allocation in SparkContext") { + val conf = createConf(0, 1, 0) + .setMaster("local-cluster[1,1,1024]") + .setAppName(getClass().getName()) + val sc0 = new SparkContext(conf) - contexts += sc0 - assert(sc0.executorAllocationManager.isDefined) - sc0.stop() + try { + assert(sc0.executorAllocationManager.isDefined) + } finally { + sc0.stop() + } + } + test("verify min/max executors") { // Min < 0 - val conf1 = conf.clone().set(config.DYN_ALLOCATION_MIN_EXECUTORS, -1) - intercept[SparkException] { contexts += new SparkContext(conf1) } + intercept[SparkException] { + createManager(createConf().set(config.DYN_ALLOCATION_MIN_EXECUTORS, -1)) + } // Max < 0 - val conf2 = conf.clone().set(config.DYN_ALLOCATION_MAX_EXECUTORS, -1) - intercept[SparkException] { contexts += new SparkContext(conf2) } + intercept[SparkException] { + createManager(createConf().set(config.DYN_ALLOCATION_MAX_EXECUTORS, -1)) + } // Both min and max, but min > max - intercept[SparkException] { createSparkContext(2, 1) } + intercept[SparkException] { + createManager(createConf(2, 1)) + } // Both min and max, and min == max - val sc1 = createSparkContext(1, 1) - assert(sc1.executorAllocationManager.isDefined) - sc1.stop() + createManager(createConf(1, 1)) // Both min and max, and min < max - val sc2 = createSparkContext(1, 2) - assert(sc2.executorAllocationManager.isDefined) - sc2.stop() + createManager(createConf(1, 2)) } test("starting state") { - sc = createSparkContext() - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf()) assert(numExecutorsTarget(manager) === 1) assert(executorsPendingToRemove(manager).isEmpty) - assert(executorIds(manager).isEmpty) assert(addTime(manager) === ExecutorAllocationManager.NOT_SET) - assert(removeTimes(manager).isEmpty) } test("add executors") { - sc = createSparkContext(1, 10, 1) - val manager = sc.executorAllocationManager.get - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 1000))) + val manager = createManager(createConf(1, 10, 1)) + post(SparkListenerStageSubmitted(createStageInfo(0, 1000))) // Keep adding until the limit is reached assert(numExecutorsTarget(manager) === 1) @@ -152,24 +154,15 @@ class ExecutorAllocationManagerSuite } def testAllocationRatio(cores: Int, divisor: Double, expected: Int): Unit = { - val conf = new SparkConf() - .setMaster("myDummyLocalExternalClusterManager") - .setAppName("test-executor-allocation-manager") - .set(config.DYN_ALLOCATION_ENABLED, true) - .set(config.DYN_ALLOCATION_TESTING, true) - .set(config.DYN_ALLOCATION_MAX_EXECUTORS, 15) - .set(config.DYN_ALLOCATION_MIN_EXECUTORS, 3) + val conf = createConf(3, 15) .set(config.DYN_ALLOCATION_EXECUTOR_ALLOCATION_RATIO, divisor) .set(config.EXECUTOR_CORES, cores) - val sc = new SparkContext(conf) - contexts += sc - var manager = sc.executorAllocationManager.get - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 20))) + val manager = createManager(conf) + post(SparkListenerStageSubmitted(createStageInfo(0, 20))) for (i <- 0 to 5) { addExecutors(manager) } assert(numExecutorsTarget(manager) === expected) - sc.stop() } test("executionAllocationRatio is correctly handled") { @@ -185,9 +178,8 @@ class ExecutorAllocationManagerSuite test("add executors capped by num pending tasks") { - sc = createSparkContext(0, 10, 0) - val manager = sc.executorAllocationManager.get - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 5))) + val manager = createManager(createConf(0, 10, 0)) + post(SparkListenerStageSubmitted(createStageInfo(0, 5))) // Verify that we're capped at number of tasks in the stage assert(numExecutorsTarget(manager) === 0) @@ -203,10 +195,10 @@ class ExecutorAllocationManagerSuite assert(numExecutorsToAdd(manager) === 1) // Verify that running a task doesn't affect the target - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(1, 3))) - post(sc.listenerBus, SparkListenerExecutorAdded( + post(SparkListenerStageSubmitted(createStageInfo(1, 3))) + post(SparkListenerExecutorAdded( 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - post(sc.listenerBus, SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-1"))) + post(SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-1"))) assert(numExecutorsTarget(manager) === 5) assert(addExecutors(manager) === 1) assert(numExecutorsTarget(manager) === 6) @@ -219,9 +211,9 @@ class ExecutorAllocationManagerSuite assert(numExecutorsToAdd(manager) === 1) // Verify that re-running a task doesn't blow things up - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(2, 3))) - post(sc.listenerBus, SparkListenerTaskStart(2, 0, createTaskInfo(0, 0, "executor-1"))) - post(sc.listenerBus, SparkListenerTaskStart(2, 0, createTaskInfo(1, 0, "executor-1"))) + post(SparkListenerStageSubmitted(createStageInfo(2, 3))) + post(SparkListenerTaskStart(2, 0, createTaskInfo(0, 0, "executor-1"))) + post(SparkListenerTaskStart(2, 0, createTaskInfo(1, 0, "executor-1"))) assert(addExecutors(manager) === 1) assert(numExecutorsTarget(manager) === 9) assert(numExecutorsToAdd(manager) === 2) @@ -230,23 +222,22 @@ class ExecutorAllocationManagerSuite assert(numExecutorsToAdd(manager) === 1) // Verify that running a task once we're at our limit doesn't blow things up - post(sc.listenerBus, SparkListenerTaskStart(2, 0, createTaskInfo(0, 1, "executor-1"))) + post(SparkListenerTaskStart(2, 0, createTaskInfo(0, 1, "executor-1"))) assert(addExecutors(manager) === 0) assert(numExecutorsTarget(manager) === 10) } test("add executors when speculative tasks added") { - sc = createSparkContext(0, 10, 0) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(0, 10, 0)) // Verify that we're capped at number of tasks including the speculative ones in the stage - post(sc.listenerBus, SparkListenerSpeculativeTaskSubmitted(1)) + post(SparkListenerSpeculativeTaskSubmitted(1)) assert(numExecutorsTarget(manager) === 0) assert(numExecutorsToAdd(manager) === 1) assert(addExecutors(manager) === 1) - post(sc.listenerBus, SparkListenerSpeculativeTaskSubmitted(1)) - post(sc.listenerBus, SparkListenerSpeculativeTaskSubmitted(1)) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(1, 2))) + post(SparkListenerSpeculativeTaskSubmitted(1)) + post(SparkListenerSpeculativeTaskSubmitted(1)) + post(SparkListenerStageSubmitted(createStageInfo(1, 2))) assert(numExecutorsTarget(manager) === 1) assert(numExecutorsToAdd(manager) === 2) assert(addExecutors(manager) === 2) @@ -257,39 +248,37 @@ class ExecutorAllocationManagerSuite assert(numExecutorsToAdd(manager) === 1) // Verify that running a task doesn't affect the target - post(sc.listenerBus, SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-1"))) + post(SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-1"))) assert(numExecutorsTarget(manager) === 5) assert(addExecutors(manager) === 0) assert(numExecutorsToAdd(manager) === 1) // Verify that running a speculative task doesn't affect the target - post(sc.listenerBus, SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-2", true))) + post(SparkListenerTaskStart(1, 0, createTaskInfo(0, 0, "executor-2", true))) assert(numExecutorsTarget(manager) === 5) assert(addExecutors(manager) === 0) assert(numExecutorsToAdd(manager) === 1) } test("ignore task end events from completed stages") { - sc = createSparkContext(0, 10, 0) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(0, 10, 0)) val stage = createStageInfo(0, 5) - post(sc.listenerBus, SparkListenerStageSubmitted(stage)) + post(SparkListenerStageSubmitted(stage)) val taskInfo1 = createTaskInfo(0, 0, "executor-1") val taskInfo2 = createTaskInfo(1, 1, "executor-1") - post(sc.listenerBus, SparkListenerTaskStart(0, 0, taskInfo1)) - post(sc.listenerBus, SparkListenerTaskStart(0, 0, taskInfo2)) + post(SparkListenerTaskStart(0, 0, taskInfo1)) + post(SparkListenerTaskStart(0, 0, taskInfo2)) - post(sc.listenerBus, SparkListenerStageCompleted(stage)) + post(SparkListenerStageCompleted(stage)) - post(sc.listenerBus, SparkListenerTaskEnd(0, 0, null, Success, taskInfo1, null)) - post(sc.listenerBus, SparkListenerTaskEnd(2, 0, null, Success, taskInfo2, null)) + post(SparkListenerTaskEnd(0, 0, null, Success, taskInfo1, null)) + post(SparkListenerTaskEnd(2, 0, null, Success, taskInfo2, null)) assert(totalRunningTasks(manager) === 0) } testRetry("cancel pending executors when no longer needed") { - sc = createSparkContext(0, 10, 0) - val manager = sc.executorAllocationManager.get - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(2, 5))) + val manager = createManager(createConf(0, 10, 0)) + post(SparkListenerStageSubmitted(createStageInfo(2, 5))) assert(numExecutorsTarget(manager) === 0) assert(numExecutorsToAdd(manager) === 1) @@ -300,26 +289,25 @@ class ExecutorAllocationManagerSuite assert(numExecutorsTarget(manager) === 3) val task1Info = createTaskInfo(0, 0, "executor-1") - post(sc.listenerBus, SparkListenerTaskStart(2, 0, task1Info)) + post(SparkListenerTaskStart(2, 0, task1Info)) assert(numExecutorsToAdd(manager) === 4) assert(addExecutors(manager) === 2) val task2Info = createTaskInfo(1, 0, "executor-1") - post(sc.listenerBus, SparkListenerTaskStart(2, 0, task2Info)) + post(SparkListenerTaskStart(2, 0, task2Info)) task1Info.markFinished(TaskState.FINISHED, System.currentTimeMillis()) - post(sc.listenerBus, SparkListenerTaskEnd(2, 0, null, Success, task1Info, null)) + post(SparkListenerTaskEnd(2, 0, null, Success, task1Info, null)) task2Info.markFinished(TaskState.FINISHED, System.currentTimeMillis()) - post(sc.listenerBus, SparkListenerTaskEnd(2, 0, null, Success, task2Info, null)) + post(SparkListenerTaskEnd(2, 0, null, Success, task2Info, null)) assert(adjustRequestedExecutors(manager) === -1) } test("remove executors") { - sc = createSparkContext(5, 10, 5) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(5, 10, 5)) (1 to 10).map(_.toString).foreach { id => onExecutorAdded(manager, id) } // Keep removing until the limit is reached @@ -332,8 +320,6 @@ class ExecutorAllocationManagerSuite assert(executorsPendingToRemove(manager).size === 3) assert(executorsPendingToRemove(manager).contains("2")) assert(executorsPendingToRemove(manager).contains("3")) - assert(!removeExecutor(manager, "100")) // remove non-existent executors - assert(!removeExecutor(manager, "101")) assert(executorsPendingToRemove(manager).size === 3) assert(removeExecutor(manager, "4")) assert(removeExecutor(manager, "5")) @@ -368,8 +354,7 @@ class ExecutorAllocationManagerSuite } test("remove multiple executors") { - sc = createSparkContext(5, 10, 5) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(5, 10, 5)) (1 to 10).map(_.toString).foreach { id => onExecutorAdded(manager, id) } // Keep removing until the limit is reached @@ -381,8 +366,6 @@ class ExecutorAllocationManagerSuite assert(executorsPendingToRemove(manager).size === 3) assert(executorsPendingToRemove(manager).contains("2")) assert(executorsPendingToRemove(manager).contains("3")) - assert(!removeExecutor(manager, "100")) // remove non-existent executors - assert(removeExecutors(manager, Seq("101", "102")) !== Seq("101", "102")) assert(executorsPendingToRemove(manager).size === 3) assert(removeExecutor(manager, "4")) assert(removeExecutors(manager, Seq("5")) === Seq("5")) @@ -417,28 +400,27 @@ class ExecutorAllocationManagerSuite } test ("Removing with various numExecutorsTarget condition") { - sc = createSparkContext(5, 12, 5) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(5, 12, 5)) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 8))) + post(SparkListenerStageSubmitted(createStageInfo(0, 8))) // Remove when numExecutorsTarget is the same as the current number of executors assert(addExecutors(manager) === 1) assert(addExecutors(manager) === 2) (1 to 8).foreach(execId => onExecutorAdded(manager, execId.toString)) (1 to 8).map { i => createTaskInfo(i, i, s"$i") }.foreach { - info => post(sc.listenerBus, SparkListenerTaskStart(0, 0, info)) } - assert(executorIds(manager).size === 8) + info => post(SparkListenerTaskStart(0, 0, info)) } + assert(manager.executorMonitor.executorCount === 8) assert(numExecutorsTarget(manager) === 8) assert(maxNumExecutorsNeeded(manager) == 8) assert(!removeExecutor(manager, "1")) // won't work since numExecutorsTarget == numExecutors // Remove executors when numExecutorsTarget is lower than current number of executors (1 to 3).map { i => createTaskInfo(i, i, s"$i") }.foreach { info => - post(sc.listenerBus, SparkListenerTaskEnd(0, 0, null, Success, info, null)) + post(SparkListenerTaskEnd(0, 0, null, Success, info, null)) } adjustRequestedExecutors(manager) - assert(executorIds(manager).size === 8) + assert(manager.executorMonitor.executorCount === 8) assert(numExecutorsTarget(manager) === 5) assert(maxNumExecutorsNeeded(manager) == 5) assert(removeExecutor(manager, "1")) @@ -448,9 +430,8 @@ class ExecutorAllocationManagerSuite onExecutorRemoved(manager, "3") // numExecutorsTarget is lower than minNumExecutors - post(sc.listenerBus, - SparkListenerTaskEnd(0, 0, null, Success, createTaskInfo(4, 4, "4"), null)) - assert(executorIds(manager).size === 5) + post(SparkListenerTaskEnd(0, 0, null, Success, createTaskInfo(4, 4, "4"), null)) + assert(manager.executorMonitor.executorCount === 5) assert(numExecutorsTarget(manager) === 5) assert(maxNumExecutorsNeeded(manager) == 4) assert(!removeExecutor(manager, "4")) // lower limit @@ -458,9 +439,8 @@ class ExecutorAllocationManagerSuite } test ("interleaving add and remove") { - sc = createSparkContext(5, 12, 5) - val manager = sc.executorAllocationManager.get - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 1000))) + val manager = createManager(createConf(5, 12, 5)) + post(SparkListenerStageSubmitted(createStageInfo(0, 1000))) // Add a few executors assert(addExecutors(manager) === 1) @@ -473,7 +453,7 @@ class ExecutorAllocationManagerSuite onExecutorAdded(manager, "6") onExecutorAdded(manager, "7") onExecutorAdded(manager, "8") - assert(executorIds(manager).size === 8) + assert(manager.executorMonitor.executorCount === 8) assert(numExecutorsTarget(manager) === 8) @@ -486,7 +466,7 @@ class ExecutorAllocationManagerSuite onExecutorAdded(manager, "10") onExecutorAdded(manager, "11") onExecutorAdded(manager, "12") - assert(executorIds(manager).size === 12) + assert(manager.executorMonitor.executorCount === 12) assert(numExecutorsTarget(manager) === 8) assert(removeExecutor(manager, "1")) @@ -497,7 +477,7 @@ class ExecutorAllocationManagerSuite onExecutorRemoved(manager, "2") onExecutorRemoved(manager, "3") onExecutorRemoved(manager, "4") - assert(executorIds(manager).size === 8) + assert(manager.executorMonitor.executorCount === 8) // Add until limit assert(!removeExecutor(manager, "7")) // still at lower limit @@ -506,34 +486,32 @@ class ExecutorAllocationManagerSuite onExecutorAdded(manager, "14") onExecutorAdded(manager, "15") onExecutorAdded(manager, "16") - assert(executorIds(manager).size === 12) + assert(manager.executorMonitor.executorCount === 12) // Remove succeeds again, now that we are no longer at the lower limit assert(removeExecutors(manager, Seq("5", "6", "7")) === Seq("5", "6", "7")) assert(removeExecutor(manager, "8")) - assert(executorIds(manager).size === 12) + assert(manager.executorMonitor.executorCount === 12) onExecutorRemoved(manager, "5") onExecutorRemoved(manager, "6") - assert(executorIds(manager).size === 10) + assert(manager.executorMonitor.executorCount === 10) assert(numExecutorsToAdd(manager) === 4) onExecutorRemoved(manager, "9") onExecutorRemoved(manager, "10") assert(addExecutors(manager) === 4) // at upper limit onExecutorAdded(manager, "17") onExecutorAdded(manager, "18") - assert(executorIds(manager).size === 10) + assert(manager.executorMonitor.executorCount === 10) assert(addExecutors(manager) === 0) // still at upper limit onExecutorAdded(manager, "19") onExecutorAdded(manager, "20") - assert(executorIds(manager).size === 12) + assert(manager.executorMonitor.executorCount === 12) assert(numExecutorsTarget(manager) === 12) } test("starting/canceling add timer") { - sc = createSparkContext(2, 10, 2) val clock = new ManualClock(8888L) - val manager = sc.executorAllocationManager.get - manager.setClock(clock) + val manager = createManager(createConf(2, 10, 2), clock = clock) // Starting add timer is idempotent assert(addTime(manager) === NOT_SET) @@ -561,58 +539,9 @@ class ExecutorAllocationManagerSuite assert(firstAddTime !== secondAddTime) } - test("starting/canceling remove timers") { - sc = createSparkContext(2, 10, 2) - val clock = new ManualClock(14444L) - val manager = sc.executorAllocationManager.get - manager.setClock(clock) - - executorIds(manager).asInstanceOf[mutable.Set[String]] ++= List("1", "2", "3") - - // Starting remove timer is idempotent for each executor - assert(removeTimes(manager).isEmpty) - onExecutorIdle(manager, "1") - assert(removeTimes(manager).size === 1) - assert(removeTimes(manager).contains("1")) - val firstRemoveTime = removeTimes(manager)("1") - assert(firstRemoveTime === clock.getTimeMillis + executorIdleTimeout * 1000) - clock.advance(100L) - onExecutorIdle(manager, "1") - assert(removeTimes(manager)("1") === firstRemoveTime) // timer is already started - clock.advance(200L) - onExecutorIdle(manager, "1") - assert(removeTimes(manager)("1") === firstRemoveTime) - clock.advance(300L) - onExecutorIdle(manager, "2") - assert(removeTimes(manager)("2") !== firstRemoveTime) // different executor - assert(removeTimes(manager)("2") === clock.getTimeMillis + executorIdleTimeout * 1000) - clock.advance(400L) - onExecutorIdle(manager, "3") - assert(removeTimes(manager)("3") !== firstRemoveTime) - assert(removeTimes(manager)("3") === clock.getTimeMillis + executorIdleTimeout * 1000) - assert(removeTimes(manager).size === 3) - assert(removeTimes(manager).contains("2")) - assert(removeTimes(manager).contains("3")) - - // Restart remove timer - clock.advance(1000L) - onExecutorBusy(manager, "1") - assert(removeTimes(manager).size === 2) - onExecutorIdle(manager, "1") - assert(removeTimes(manager).size === 3) - assert(removeTimes(manager).contains("1")) - val secondRemoveTime = removeTimes(manager)("1") - assert(secondRemoveTime === clock.getTimeMillis + executorIdleTimeout * 1000) - assert(removeTimes(manager)("1") === secondRemoveTime) // timer is already started - assert(removeTimes(manager)("1") !== firstRemoveTime) - assert(firstRemoveTime !== secondRemoveTime) - } - test("mock polling loop with no events") { - sc = createSparkContext(0, 20, 0) - val manager = sc.executorAllocationManager.get val clock = new ManualClock(2020L) - manager.setClock(clock) + val manager = createManager(createConf(0, 20, 0), clock = clock) // No events - we should not be adding or removing assert(numExecutorsTarget(manager) === 0) @@ -635,11 +564,9 @@ class ExecutorAllocationManagerSuite } test("mock polling loop add behavior") { - sc = createSparkContext(0, 20, 0) val clock = new ManualClock(2020L) - val manager = sc.executorAllocationManager.get - manager.setClock(clock) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 1000))) + val manager = createManager(createConf(0, 20, 0), clock = clock) + post(SparkListenerStageSubmitted(createStageInfo(0, 1000))) // Scheduler queue backlogged onSchedulerBacklogged(manager) @@ -685,24 +612,25 @@ class ExecutorAllocationManagerSuite } test("mock polling loop remove behavior") { - sc = createSparkContext(1, 20, 1) val clock = new ManualClock(2020L) - val manager = sc.executorAllocationManager.get - manager.setClock(clock) + val manager = createManager(createConf(1, 20, 1), clock = clock) // Remove idle executors on timeout onExecutorAdded(manager, "executor-1") onExecutorAdded(manager, "executor-2") onExecutorAdded(manager, "executor-3") - assert(removeTimes(manager).size === 3) assert(executorsPendingToRemove(manager).isEmpty) + + // idle threshold not reached yet clock.advance(executorIdleTimeout * 1000 / 2) schedule(manager) - assert(removeTimes(manager).size === 3) // idle threshold not reached yet + assert(manager.executorMonitor.timedOutExecutors().isEmpty) assert(executorsPendingToRemove(manager).isEmpty) + + // idle threshold exceeded clock.advance(executorIdleTimeout * 1000) + assert(manager.executorMonitor.timedOutExecutors().size === 3) schedule(manager) - assert(removeTimes(manager).isEmpty) // idle threshold exceeded assert(executorsPendingToRemove(manager).size === 2) // limit reached (1 executor remaining) // Mark a subset as busy - only idle executors should be removed @@ -710,20 +638,20 @@ class ExecutorAllocationManagerSuite onExecutorAdded(manager, "executor-5") onExecutorAdded(manager, "executor-6") onExecutorAdded(manager, "executor-7") - assert(removeTimes(manager).size === 5) // 5 active executors + assert(manager.executorMonitor.executorCount === 7) assert(executorsPendingToRemove(manager).size === 2) // 2 pending to be removed onExecutorBusy(manager, "executor-4") onExecutorBusy(manager, "executor-5") onExecutorBusy(manager, "executor-6") // 3 busy and 2 idle (of the 5 active ones) + + // after scheduling, the previously timed out executor should be removed, since + // there are new active ones. schedule(manager) - assert(removeTimes(manager).size === 2) // remove only idle executors - assert(!removeTimes(manager).contains("executor-4")) - assert(!removeTimes(manager).contains("executor-5")) - assert(!removeTimes(manager).contains("executor-6")) - assert(executorsPendingToRemove(manager).size === 2) + assert(executorsPendingToRemove(manager).size === 3) + + // advance the clock so that idle executors should time out and move to the pending list clock.advance(executorIdleTimeout * 1000) schedule(manager) - assert(removeTimes(manager).isEmpty) // idle executors are removed assert(executorsPendingToRemove(manager).size === 4) assert(!executorsPendingToRemove(manager).contains("executor-4")) assert(!executorsPendingToRemove(manager).contains("executor-5")) @@ -734,152 +662,45 @@ class ExecutorAllocationManagerSuite onExecutorIdle(manager, "executor-5") onExecutorIdle(manager, "executor-6") schedule(manager) - assert(removeTimes(manager).size === 3) // 0 busy and 3 idle - assert(removeTimes(manager).contains("executor-4")) - assert(removeTimes(manager).contains("executor-5")) - assert(removeTimes(manager).contains("executor-6")) assert(executorsPendingToRemove(manager).size === 4) clock.advance(executorIdleTimeout * 1000) schedule(manager) - assert(removeTimes(manager).isEmpty) assert(executorsPendingToRemove(manager).size === 6) // limit reached (1 executor remaining) } test("listeners trigger add executors correctly") { - sc = createSparkContext(2, 10, 2) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(1, 20, 1)) assert(addTime(manager) === NOT_SET) // Starting a stage should start the add timer val numTasks = 10 - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, numTasks))) + post(SparkListenerStageSubmitted(createStageInfo(0, numTasks))) assert(addTime(manager) !== NOT_SET) // Starting a subset of the tasks should not cancel the add timer val taskInfos = (0 to numTasks - 1).map { i => createTaskInfo(i, i, "executor-1") } - taskInfos.tail.foreach { info => post(sc.listenerBus, SparkListenerTaskStart(0, 0, info)) } + taskInfos.tail.foreach { info => post(SparkListenerTaskStart(0, 0, info)) } assert(addTime(manager) !== NOT_SET) // Starting all remaining tasks should cancel the add timer - post(sc.listenerBus, SparkListenerTaskStart(0, 0, taskInfos.head)) + post(SparkListenerTaskStart(0, 0, taskInfos.head)) assert(addTime(manager) === NOT_SET) // Start two different stages // The add timer should be canceled only if all tasks in both stages start running - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(1, numTasks))) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(2, numTasks))) + post(SparkListenerStageSubmitted(createStageInfo(1, numTasks))) + post(SparkListenerStageSubmitted(createStageInfo(2, numTasks))) assert(addTime(manager) !== NOT_SET) - taskInfos.foreach { info => post(sc.listenerBus, SparkListenerTaskStart(1, 0, info)) } + taskInfos.foreach { info => post(SparkListenerTaskStart(1, 0, info)) } assert(addTime(manager) !== NOT_SET) - taskInfos.foreach { info => post(sc.listenerBus, SparkListenerTaskStart(2, 0, info)) } + taskInfos.foreach { info => post(SparkListenerTaskStart(2, 0, info)) } assert(addTime(manager) === NOT_SET) } - test("listeners trigger remove executors correctly") { - sc = createSparkContext(2, 10, 2) - val manager = sc.executorAllocationManager.get - assert(removeTimes(manager).isEmpty) - - // Added executors should start the remove timers for each executor - (1 to 5).map("executor-" + _).foreach { id => onExecutorAdded(manager, id) } - assert(removeTimes(manager).size === 5) - - // Starting a task cancel the remove timer for that executor - post(sc.listenerBus, SparkListenerTaskStart(0, 0, createTaskInfo(0, 0, "executor-1"))) - post(sc.listenerBus, SparkListenerTaskStart(0, 0, createTaskInfo(1, 1, "executor-1"))) - post(sc.listenerBus, SparkListenerTaskStart(0, 0, createTaskInfo(2, 2, "executor-2"))) - assert(removeTimes(manager).size === 3) - assert(!removeTimes(manager).contains("executor-1")) - assert(!removeTimes(manager).contains("executor-2")) - - // Finishing all tasks running on an executor should start the remove timer for that executor - post(sc.listenerBus, SparkListenerTaskEnd( - 0, 0, "task-type", Success, createTaskInfo(0, 0, "executor-1"), new TaskMetrics)) - post(sc.listenerBus, SparkListenerTaskEnd( - 0, 0, "task-type", Success, createTaskInfo(2, 2, "executor-2"), new TaskMetrics)) - assert(removeTimes(manager).size === 4) - assert(!removeTimes(manager).contains("executor-1")) // executor-1 has not finished yet - assert(removeTimes(manager).contains("executor-2")) - post(sc.listenerBus, SparkListenerTaskEnd( - 0, 0, "task-type", Success, createTaskInfo(1, 1, "executor-1"), new TaskMetrics)) - assert(removeTimes(manager).size === 5) - assert(removeTimes(manager).contains("executor-1")) // executor-1 has now finished - } - - test("listeners trigger add and remove executor callbacks correctly") { - sc = createSparkContext(2, 10, 2) - val manager = sc.executorAllocationManager.get - assert(executorIds(manager).isEmpty) - assert(removeTimes(manager).isEmpty) - - // New executors have registered - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - assert(executorIds(manager).size === 1) - assert(executorIds(manager).contains("executor-1")) - assert(removeTimes(manager).size === 1) - assert(removeTimes(manager).contains("executor-1")) - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "executor-2", new ExecutorInfo("host2", 1, Map.empty, Map.empty))) - assert(executorIds(manager).size === 2) - assert(executorIds(manager).contains("executor-2")) - assert(removeTimes(manager).size === 2) - assert(removeTimes(manager).contains("executor-2")) - - // Existing executors have disconnected - post(sc.listenerBus, SparkListenerExecutorRemoved(0L, "executor-1", "")) - assert(executorIds(manager).size === 1) - assert(!executorIds(manager).contains("executor-1")) - assert(removeTimes(manager).size === 1) - assert(!removeTimes(manager).contains("executor-1")) - - // Unknown executor has disconnected - post(sc.listenerBus, SparkListenerExecutorRemoved(0L, "executor-3", "")) - assert(executorIds(manager).size === 1) - assert(removeTimes(manager).size === 1) - } - - test("SPARK-4951: call onTaskStart before onExecutorAdded") { - sc = createSparkContext(2, 10, 2) - val manager = sc.executorAllocationManager.get - assert(executorIds(manager).isEmpty) - assert(removeTimes(manager).isEmpty) - - post(sc.listenerBus, SparkListenerTaskStart(0, 0, createTaskInfo(0, 0, "executor-1"))) - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - assert(executorIds(manager).size === 1) - assert(executorIds(manager).contains("executor-1")) - assert(removeTimes(manager).size === 0) - } - - test("SPARK-4951: onExecutorAdded should not add a busy executor to removeTimes") { - sc = createSparkContext(2, 10) - val manager = sc.executorAllocationManager.get - assert(executorIds(manager).isEmpty) - assert(removeTimes(manager).isEmpty) - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "executor-1", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - post(sc.listenerBus, SparkListenerTaskStart(0, 0, createTaskInfo(0, 0, "executor-1"))) - - assert(executorIds(manager).size === 1) - assert(executorIds(manager).contains("executor-1")) - assert(removeTimes(manager).size === 0) - - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "executor-2", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - assert(executorIds(manager).size === 2) - assert(executorIds(manager).contains("executor-2")) - assert(removeTimes(manager).size === 1) - assert(removeTimes(manager).contains("executor-2")) - assert(!removeTimes(manager).contains("executor-1")) - } - test("avoid ramp up when target < running executors") { - sc = createSparkContext(0, 100000, 0) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(0, 100000, 0)) val stage1 = createStageInfo(0, 1000) - post(sc.listenerBus, SparkListenerStageSubmitted(stage1)) + post(SparkListenerStageSubmitted(stage1)) assert(addExecutors(manager) === 1) assert(addExecutors(manager) === 2) @@ -889,22 +710,20 @@ class ExecutorAllocationManagerSuite (0 until 15).foreach { i => onExecutorAdded(manager, s"executor-$i") } - assert(executorIds(manager).size === 15) - post(sc.listenerBus, SparkListenerStageCompleted(stage1)) + assert(manager.executorMonitor.executorCount === 15) + post(SparkListenerStageCompleted(stage1)) adjustRequestedExecutors(manager) assert(numExecutorsTarget(manager) === 0) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(1, 1000))) + post(SparkListenerStageSubmitted(createStageInfo(1, 1000))) addExecutors(manager) assert(numExecutorsTarget(manager) === 16) } test("avoid ramp down initial executors until first job is submitted") { - sc = createSparkContext(2, 5, 3) - val manager = sc.executorAllocationManager.get val clock = new ManualClock(10000L) - manager.setClock(clock) + val manager = createManager(createConf(2, 5, 3), clock = clock) // Verify the initial number of executors assert(numExecutorsTarget(manager) === 3) @@ -912,7 +731,7 @@ class ExecutorAllocationManagerSuite // Verify whether the initial number of executors is kept with no pending tasks assert(numExecutorsTarget(manager) === 3) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(1, 2))) + post(SparkListenerStageSubmitted(createStageInfo(1, 2))) clock.advance(100L) assert(maxNumExecutorsNeeded(manager) === 2) @@ -923,10 +742,8 @@ class ExecutorAllocationManagerSuite } test("avoid ramp down initial executors until idle executor is timeout") { - sc = createSparkContext(2, 5, 3) - val manager = sc.executorAllocationManager.get val clock = new ManualClock(10000L) - manager.setClock(clock) + val manager = createManager(createConf(2, 5, 3), clock = clock) // Verify the initial number of executors assert(numExecutorsTarget(manager) === 3) @@ -946,8 +763,7 @@ class ExecutorAllocationManagerSuite } test("get pending task number and related locality preference") { - sc = createSparkContext(2, 5, 3) - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf(2, 5, 3)) val localityPreferences1 = Seq( Seq(TaskLocation("host1"), TaskLocation("host2"), TaskLocation("host3")), @@ -957,7 +773,7 @@ class ExecutorAllocationManagerSuite Seq.empty ) val stageInfo1 = createStageInfo(1, 5, localityPreferences1) - post(sc.listenerBus, SparkListenerStageSubmitted(stageInfo1)) + post(SparkListenerStageSubmitted(stageInfo1)) assert(localityAwareTasks(manager) === 3) assert(hostToLocalTaskCount(manager) === @@ -969,45 +785,43 @@ class ExecutorAllocationManagerSuite Seq.empty ) val stageInfo2 = createStageInfo(2, 3, localityPreferences2) - post(sc.listenerBus, SparkListenerStageSubmitted(stageInfo2)) + post(SparkListenerStageSubmitted(stageInfo2)) assert(localityAwareTasks(manager) === 5) assert(hostToLocalTaskCount(manager) === Map("host1" -> 2, "host2" -> 4, "host3" -> 4, "host4" -> 3, "host5" -> 2)) - post(sc.listenerBus, SparkListenerStageCompleted(stageInfo1)) + post(SparkListenerStageCompleted(stageInfo1)) assert(localityAwareTasks(manager) === 2) assert(hostToLocalTaskCount(manager) === Map("host2" -> 1, "host3" -> 2, "host4" -> 1, "host5" -> 2)) } test("SPARK-8366: maxNumExecutorsNeeded should properly handle failed tasks") { - sc = createSparkContext() - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf()) assert(maxNumExecutorsNeeded(manager) === 0) - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 1))) + post(SparkListenerStageSubmitted(createStageInfo(0, 1))) assert(maxNumExecutorsNeeded(manager) === 1) val taskInfo = createTaskInfo(1, 1, "executor-1") - post(sc.listenerBus, SparkListenerTaskStart(0, 0, taskInfo)) + post(SparkListenerTaskStart(0, 0, taskInfo)) assert(maxNumExecutorsNeeded(manager) === 1) // If the task is failed, we expect it to be resubmitted later. val taskEndReason = ExceptionFailure(null, null, null, null, None) - post(sc.listenerBus, SparkListenerTaskEnd(0, 0, null, taskEndReason, taskInfo, null)) + post(SparkListenerTaskEnd(0, 0, null, taskEndReason, taskInfo, null)) assert(maxNumExecutorsNeeded(manager) === 1) } test("reset the state of allocation manager") { - sc = createSparkContext() - val manager = sc.executorAllocationManager.get + val manager = createManager(createConf()) assert(numExecutorsTarget(manager) === 1) assert(numExecutorsToAdd(manager) === 1) // Allocation manager is reset when adding executor requests are sent without reporting back // executor added. - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 10))) + post(SparkListenerStageSubmitted(createStageInfo(0, 10))) assert(addExecutors(manager) === 1) assert(numExecutorsTarget(manager) === 2) @@ -1019,10 +833,10 @@ class ExecutorAllocationManagerSuite manager.reset() assert(numExecutorsTarget(manager) === 1) assert(numExecutorsToAdd(manager) === 1) - assert(executorIds(manager) === Set.empty) + assert(manager.executorMonitor.executorCount === 0) // Allocation manager is reset when executors are added. - post(sc.listenerBus, SparkListenerStageSubmitted(createStageInfo(0, 10))) + post(SparkListenerStageSubmitted(createStageInfo(0, 10))) addExecutors(manager) addExecutors(manager) @@ -1034,7 +848,7 @@ class ExecutorAllocationManagerSuite onExecutorAdded(manager, "third") onExecutorAdded(manager, "fourth") onExecutorAdded(manager, "fifth") - assert(executorIds(manager) === Set("first", "second", "third", "fourth", "fifth")) + assert(manager.executorMonitor.executorCount === 5) // Cluster manager lost will make all the live executors lost, so here simulate this behavior onExecutorRemoved(manager, "first") @@ -1046,8 +860,7 @@ class ExecutorAllocationManagerSuite manager.reset() assert(numExecutorsTarget(manager) === 1) assert(numExecutorsToAdd(manager) === 1) - assert(executorIds(manager) === Set.empty) - assert(removeTimes(manager) === Map.empty) + assert(manager.executorMonitor.executorCount === 0) // Allocation manager is reset when executors are pending to remove addExecutors(manager) @@ -1063,14 +876,12 @@ class ExecutorAllocationManagerSuite onExecutorAdded(manager, "sixth") onExecutorAdded(manager, "seventh") onExecutorAdded(manager, "eighth") - assert(executorIds(manager) === Set("first", "second", "third", "fourth", "fifth", - "sixth", "seventh", "eighth")) + assert(manager.executorMonitor.executorCount === 8) removeExecutor(manager, "first") removeExecutors(manager, Seq("second", "third")) assert(executorsPendingToRemove(manager) === Set("first", "second", "third")) - assert(executorIds(manager) === Set("first", "second", "third", "fourth", "fifth", - "sixth", "seventh", "eighth")) + assert(manager.executorMonitor.executorCount === 8) // Cluster manager lost will make all the live executors lost, so here simulate this behavior @@ -1085,80 +896,64 @@ class ExecutorAllocationManagerSuite assert(numExecutorsTarget(manager) === 1) assert(numExecutorsToAdd(manager) === 1) assert(executorsPendingToRemove(manager) === Set.empty) - assert(removeTimes(manager) === Map.empty) + assert(manager.executorMonitor.executorCount === 0) } test("SPARK-23365 Don't update target num executors when killing idle executors") { - val minExecutors = 1 - val initialExecutors = 1 - val maxExecutors = 2 - val conf = new SparkConf() - .set(config.DYN_ALLOCATION_ENABLED, true) - .set(config.SHUFFLE_SERVICE_ENABLED, true) - .set(config.DYN_ALLOCATION_MIN_EXECUTORS, minExecutors) - .set(config.DYN_ALLOCATION_MAX_EXECUTORS, maxExecutors) - .set(config.DYN_ALLOCATION_INITIAL_EXECUTORS, initialExecutors) - .set(config.DYN_ALLOCATION_SCHEDULER_BACKLOG_TIMEOUT.key, "1000ms") - .set(config.DYN_ALLOCATION_SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT.key, "1000ms") - .set(config.DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT.key, "3000ms") - val mockAllocationClient = mock(classOf[ExecutorAllocationClient]) - val mockBMM = mock(classOf[BlockManagerMaster]) - val manager = new ExecutorAllocationManager( - mockAllocationClient, mock(classOf[LiveListenerBus]), conf, mockBMM) val clock = new ManualClock() - manager.setClock(clock) + val manager = createManager( + createConf(1, 2, 1).set(config.DYN_ALLOCATION_TESTING, false), + clock = clock) - when(mockAllocationClient.requestTotalExecutors(meq(2), any(), any())).thenReturn(true) + when(client.requestTotalExecutors(meq(2), any(), any())).thenReturn(true) // test setup -- job with 2 tasks, scale up to two executors assert(numExecutorsTarget(manager) === 1) - manager.listener.onExecutorAdded(SparkListenerExecutorAdded( + post(SparkListenerExecutorAdded( clock.getTimeMillis(), "executor-1", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - manager.listener.onStageSubmitted(SparkListenerStageSubmitted(createStageInfo(0, 2))) + post(SparkListenerStageSubmitted(createStageInfo(0, 2))) clock.advance(1000) manager invokePrivate _updateAndSyncNumExecutorsTarget(clock.getTimeMillis()) assert(numExecutorsTarget(manager) === 2) val taskInfo0 = createTaskInfo(0, 0, "executor-1") - manager.listener.onTaskStart(SparkListenerTaskStart(0, 0, taskInfo0)) - manager.listener.onExecutorAdded(SparkListenerExecutorAdded( + post(SparkListenerTaskStart(0, 0, taskInfo0)) + post(SparkListenerExecutorAdded( clock.getTimeMillis(), "executor-2", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) val taskInfo1 = createTaskInfo(1, 1, "executor-2") - manager.listener.onTaskStart(SparkListenerTaskStart(0, 0, taskInfo1)) + post(SparkListenerTaskStart(0, 0, taskInfo1)) assert(numExecutorsTarget(manager) === 2) // have one task finish -- we should adjust the target number of executors down // but we should *not* kill any executors yet - manager.listener.onTaskEnd(SparkListenerTaskEnd(0, 0, null, Success, taskInfo0, null)) + post(SparkListenerTaskEnd(0, 0, null, Success, taskInfo0, null)) assert(maxNumExecutorsNeeded(manager) === 1) assert(numExecutorsTarget(manager) === 2) clock.advance(1000) manager invokePrivate _updateAndSyncNumExecutorsTarget(clock.getTimeMillis()) assert(numExecutorsTarget(manager) === 1) - verify(mockAllocationClient, never).killExecutors(any(), any(), any(), any()) + verify(client, never).killExecutors(any(), any(), any(), any()) // now we cross the idle timeout for executor-1, so we kill it. the really important // thing here is that we do *not* ask the executor allocation client to adjust the target // number of executors down - when(mockAllocationClient.killExecutors(Seq("executor-1"), false, false, false)) + when(client.killExecutors(Seq("executor-1"), false, false, false)) .thenReturn(Seq("executor-1")) clock.advance(3000) schedule(manager) assert(maxNumExecutorsNeeded(manager) === 1) assert(numExecutorsTarget(manager) === 1) // here's the important verify -- we did kill the executors, but did not adjust the target count - verify(mockAllocationClient).killExecutors(Seq("executor-1"), false, false, false) + verify(client).killExecutors(Seq("executor-1"), false, false, false) } test("SPARK-26758 check executor target number after idle time out ") { - sc = createSparkContext(1, 5, 3) - val manager = sc.executorAllocationManager.get val clock = new ManualClock(10000L) - manager.setClock(clock) + val manager = createManager(createConf(1, 5, 3), clock = clock) assert(numExecutorsTarget(manager) === 3) - manager.listener.onExecutorAdded(SparkListenerExecutorAdded( + post(SparkListenerExecutorAdded( clock.getTimeMillis(), "executor-1", new ExecutorInfo("host1", 1, Map.empty))) - manager.listener.onExecutorAdded(SparkListenerExecutorAdded( + post(SparkListenerExecutorAdded( clock.getTimeMillis(), "executor-2", new ExecutorInfo("host1", 2, Map.empty))) - manager.listener.onExecutorAdded(SparkListenerExecutorAdded( + post(SparkListenerExecutorAdded( clock.getTimeMillis(), "executor-3", new ExecutorInfo("host1", 3, Map.empty))) // make all the executors as idle, so that it will be killed clock.advance(executorIdleTimeout * 1000) @@ -1167,36 +962,11 @@ class ExecutorAllocationManagerSuite assert(numExecutorsTarget(manager) === 1) } - test("SPARK-26927 call onExecutorRemoved before onTaskStart") { - sc = createSparkContext(2, 5) - val manager = sc.executorAllocationManager.get - assert(executorIds(manager).isEmpty) - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "1", new ExecutorInfo("host1", 1, Map.empty, Map.empty))) - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "2", new ExecutorInfo("host2", 1, Map.empty, Map.empty))) - post(sc.listenerBus, SparkListenerExecutorAdded( - 0L, "3", new ExecutorInfo("host3", 1, Map.empty, Map.empty))) - assert(executorIds(manager).size === 3) - - post(sc.listenerBus, SparkListenerExecutorRemoved(0L, "3", "disconnected")) - assert(executorIds(manager).size === 2) - assert(executorIds(manager) === Set("1", "2")) - - val taskInfo1 = createTaskInfo(0, 0, "3") - post(sc.listenerBus, SparkListenerTaskStart(0, 0, taskInfo1)) - // Verify taskStart not adding already removed executors. - assert(executorIds(manager).size === 2) - assert(executorIds(manager) === Set("1", "2")) - } - - private def createSparkContext( + private def createConf( minExecutors: Int = 1, maxExecutors: Int = 5, - initialExecutors: Int = 1): SparkContext = { - val conf = new SparkConf() - .setMaster("myDummyLocalExternalClusterManager") - .setAppName("test-executor-allocation-manager") + initialExecutors: Int = 1): SparkConf = { + new SparkConf() .set(config.DYN_ALLOCATION_ENABLED, true) .set(config.DYN_ALLOCATION_MIN_EXECUTORS, minExecutors) .set(config.DYN_ALLOCATION_MAX_EXECUTORS, maxExecutors) @@ -1206,15 +976,49 @@ class ExecutorAllocationManagerSuite .set(config.DYN_ALLOCATION_SUSTAINED_SCHEDULER_BACKLOG_TIMEOUT.key, s"${sustainedSchedulerBacklogTimeout.toString}s") .set(config.DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT.key, s"${executorIdleTimeout.toString}s") + .set(config.SHUFFLE_SERVICE_ENABLED, true) .set(config.DYN_ALLOCATION_TESTING, true) // SPARK-22864: effectively disable the allocation schedule by setting the period to a // really long value. .set(TEST_SCHEDULE_INTERVAL, 10000L) - val sc = new SparkContext(conf) - contexts += sc - sc } + private def createManager( + conf: SparkConf, + clock: Clock = new SystemClock()): ExecutorAllocationManager = { + val manager = new ExecutorAllocationManager(client, listenerBus, conf, clock) + managers += manager + manager.start() + manager + } + + private def onExecutorAdded(manager: ExecutorAllocationManager, id: String): Unit = { + post(SparkListenerExecutorAdded(0L, id, null)) + } + + private def onExecutorRemoved(manager: ExecutorAllocationManager, id: String): Unit = { + post(SparkListenerExecutorRemoved(0L, id, null)) + } + + private def onExecutorBusy(manager: ExecutorAllocationManager, id: String): Unit = { + val info = new TaskInfo(1, 1, 1, 0, id, "foo.example.com", TaskLocality.PROCESS_LOCAL, false) + post(SparkListenerTaskStart(1, 1, info)) + } + + private def onExecutorIdle(manager: ExecutorAllocationManager, id: String): Unit = { + val info = new TaskInfo(1, 1, 1, 0, id, "foo.example.com", TaskLocality.PROCESS_LOCAL, false) + info.markFinished(TaskState.FINISHED, 1) + post(SparkListenerTaskEnd(1, 1, "foo", Success, info, null)) + } + + private def removeExecutor(manager: ExecutorAllocationManager, executorId: String): Boolean = { + val executorsRemoved = removeExecutors(manager, Seq(executorId)) + executorsRemoved.nonEmpty && executorsRemoved(0) == executorId + } + + private def executorsPendingToRemove(manager: ExecutorAllocationManager): Set[String] = { + manager.executorMonitor.executorsPendingToRemove() + } } /** @@ -1243,7 +1047,6 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester { new TaskInfo(taskId, taskIndex, 0, 0, executorId, "", TaskLocality.ANY, speculative) } - /* ------------------------------------------------------- * | Helper methods for accessing private methods and fields | * ------------------------------------------------------- */ @@ -1251,23 +1054,14 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester { private val _numExecutorsToAdd = PrivateMethod[Int]('numExecutorsToAdd) private val _numExecutorsTarget = PrivateMethod[Int]('numExecutorsTarget) private val _maxNumExecutorsNeeded = PrivateMethod[Int]('maxNumExecutorsNeeded) - private val _executorsPendingToRemove = - PrivateMethod[collection.Set[String]]('executorsPendingToRemove) - private val _executorIds = PrivateMethod[collection.Set[String]]('executorIds) private val _addTime = PrivateMethod[Long]('addTime) - private val _removeTimes = PrivateMethod[collection.Map[String, Long]]('removeTimes) private val _schedule = PrivateMethod[Unit]('schedule) private val _addExecutors = PrivateMethod[Int]('addExecutors) private val _updateAndSyncNumExecutorsTarget = PrivateMethod[Int]('updateAndSyncNumExecutorsTarget) - private val _removeExecutor = PrivateMethod[Boolean]('removeExecutor) private val _removeExecutors = PrivateMethod[Seq[String]]('removeExecutors) - private val _onExecutorAdded = PrivateMethod[Unit]('onExecutorAdded) - private val _onExecutorRemoved = PrivateMethod[Unit]('onExecutorRemoved) private val _onSchedulerBacklogged = PrivateMethod[Unit]('onSchedulerBacklogged) private val _onSchedulerQueueEmpty = PrivateMethod[Unit]('onSchedulerQueueEmpty) - private val _onExecutorIdle = PrivateMethod[Unit]('onExecutorIdle) - private val _onExecutorBusy = PrivateMethod[Unit]('onExecutorBusy) private val _localityAwareTasks = PrivateMethod[Int]('localityAwareTasks) private val _hostToLocalTaskCount = PrivateMethod[Map[String, Int]]('hostToLocalTaskCount) private val _onSpeculativeTaskSubmitted = PrivateMethod[Unit]('onSpeculativeTaskSubmitted) @@ -1281,23 +1075,10 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester { manager invokePrivate _numExecutorsTarget() } - private def executorsPendingToRemove( - manager: ExecutorAllocationManager): collection.Set[String] = { - manager invokePrivate _executorsPendingToRemove() - } - - private def executorIds(manager: ExecutorAllocationManager): collection.Set[String] = { - manager invokePrivate _executorIds() - } - private def addTime(manager: ExecutorAllocationManager): Long = { manager invokePrivate _addTime() } - private def removeTimes(manager: ExecutorAllocationManager): collection.Map[String, Long] = { - manager invokePrivate _removeTimes() - } - private def schedule(manager: ExecutorAllocationManager): Unit = { manager invokePrivate _schedule() } @@ -1315,22 +1096,10 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester { manager invokePrivate _updateAndSyncNumExecutorsTarget(0L) } - private def removeExecutor(manager: ExecutorAllocationManager, id: String): Boolean = { - manager invokePrivate _removeExecutor(id) - } - private def removeExecutors(manager: ExecutorAllocationManager, ids: Seq[String]): Seq[String] = { manager invokePrivate _removeExecutors(ids) } - private def onExecutorAdded(manager: ExecutorAllocationManager, id: String): Unit = { - manager invokePrivate _onExecutorAdded(id) - } - - private def onExecutorRemoved(manager: ExecutorAllocationManager, id: String): Unit = { - manager invokePrivate _onExecutorRemoved(id) - } - private def onSchedulerBacklogged(manager: ExecutorAllocationManager): Unit = { manager invokePrivate _onSchedulerBacklogged() } @@ -1339,14 +1108,6 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester { manager invokePrivate _onSchedulerQueueEmpty() } - private def onExecutorIdle(manager: ExecutorAllocationManager, id: String): Unit = { - manager invokePrivate _onExecutorIdle(id) - } - - private def onExecutorBusy(manager: ExecutorAllocationManager, id: String): Unit = { - manager invokePrivate _onExecutorBusy(id) - } - private def onSpeculativeTaskSubmitted(manager: ExecutorAllocationManager, id: String) : Unit = { manager invokePrivate _onSpeculativeTaskSubmitted(id) } @@ -1363,66 +1124,3 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester { manager invokePrivate _hostToLocalTaskCount() } } - -/** - * A cluster manager which wraps around the scheduler and backend for local mode. It is used for - * testing the dynamic allocation policy. - */ -private class DummyLocalExternalClusterManager extends ExternalClusterManager { - - def canCreate(masterURL: String): Boolean = masterURL == "myDummyLocalExternalClusterManager" - - override def createTaskScheduler( - sc: SparkContext, - masterURL: String): TaskScheduler = new TaskSchedulerImpl(sc, 1, isLocal = true) - - override def createSchedulerBackend( - sc: SparkContext, - masterURL: String, - scheduler: TaskScheduler): SchedulerBackend = { - val sb = new LocalSchedulerBackend(sc.getConf, scheduler.asInstanceOf[TaskSchedulerImpl], 1) - new DummyLocalSchedulerBackend(sc, sb) - } - - override def initialize(scheduler: TaskScheduler, backend: SchedulerBackend): Unit = { - val sc = scheduler.asInstanceOf[TaskSchedulerImpl] - sc.initialize(backend) - } -} - -/** - * A scheduler backend which wraps around local scheduler backend and exposes the executor - * allocation client interface for testing dynamic allocation. - */ -private class DummyLocalSchedulerBackend (sc: SparkContext, sb: SchedulerBackend) - extends SchedulerBackend with ExecutorAllocationClient { - - override private[spark] def getExecutorIds(): Seq[String] = Nil - - override private[spark] def requestTotalExecutors( - numExecutors: Int, - localityAwareTasks: Int, - hostToLocalTaskCount: Map[String, Int]): Boolean = true - - override def requestExecutors(numAdditionalExecutors: Int): Boolean = true - - override def killExecutors( - executorIds: Seq[String], - adjustTargetNumExecutors: Boolean, - countFailures: Boolean, - force: Boolean): Seq[String] = executorIds - - override def start(): Unit = sb.start() - - override def stop(): Unit = sb.stop() - - override def reviveOffers(): Unit = sb.reviveOffers() - - override def defaultParallelism(): Int = sb.defaultParallelism() - - override def maxNumConcurrentTasks(): Int = sb.maxNumConcurrentTasks() - - override def killExecutorsOnHost(host: String): Boolean = { - false - } -} diff --git a/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceDbSuite.scala b/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceDbSuite.scala index e33c3f8..e38951f 100644 --- a/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceDbSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/ExternalShuffleServiceDbSuite.scala @@ -23,6 +23,7 @@ import java.nio.charset.StandardCharsets import com.google.common.io.CharStreams import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite} +import org.apache.spark.internal.config._ import org.apache.spark.network.shuffle.{ExternalShuffleBlockHandler, ExternalShuffleBlockResolver} import org.apache.spark.network.shuffle.TestShuffleDataContext import org.apache.spark.util.Utils @@ -70,10 +71,12 @@ class ExternalShuffleServiceDbSuite extends SparkFunSuite { } } + def shuffleServiceConf: SparkConf = sparkConf.clone().set(SHUFFLE_SERVICE_PORT, 0) + def registerExecutor(): Unit = { try { sparkConf.set("spark.shuffle.service.db.enabled", "true") - externalShuffleService = new ExternalShuffleService(sparkConf, securityManager) + externalShuffleService = new ExternalShuffleService(shuffleServiceConf, securityManager) // external Shuffle Service start externalShuffleService.start() @@ -94,7 +97,7 @@ class ExternalShuffleServiceDbSuite extends SparkFunSuite { "shuffle service restart") { try { sparkConf.set("spark.shuffle.service.db.enabled", "true") - externalShuffleService = new ExternalShuffleService(sparkConf, securityManager) + externalShuffleService = new ExternalShuffleService(shuffleServiceConf, securityManager) // externalShuffleService restart externalShuffleService.start() blockHandler = externalShuffleService.getBlockHandler @@ -120,7 +123,7 @@ class ExternalShuffleServiceDbSuite extends SparkFunSuite { " shuffle service restart") { try { sparkConf.set("spark.shuffle.service.db.enabled", "false") - externalShuffleService = new ExternalShuffleService(sparkConf, securityManager) + externalShuffleService = new ExternalShuffleService(shuffleServiceConf, securityManager) // externalShuffleService restart externalShuffleService.start() blockHandler = externalShuffleService.getBlockHandler diff --git a/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala new file mode 100644 index 0000000..8d1577e --- /dev/null +++ b/core/src/test/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitorSuite.scala @@ -0,0 +1,289 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.scheduler.dynalloc + +import java.util.concurrent.TimeUnit + +import scala.collection.mutable + +import org.mockito.ArgumentMatchers.any +import org.mockito.Mockito.{mock, when} + +import org.apache.spark._ +import org.apache.spark.internal.config._ +import org.apache.spark.scheduler._ +import org.apache.spark.storage._ +import org.apache.spark.util.ManualClock + +class ExecutorMonitorSuite extends SparkFunSuite { + + private val idleTimeoutMs = TimeUnit.SECONDS.toMillis(60L) + private val storageTimeoutMs = TimeUnit.SECONDS.toMillis(120L) + + private val conf = new SparkConf() + .set(DYN_ALLOCATION_EXECUTOR_IDLE_TIMEOUT.key, "60s") + .set(DYN_ALLOCATION_CACHED_EXECUTOR_IDLE_TIMEOUT.key, "120s") + + private var monitor: ExecutorMonitor = _ + private var client: ExecutorAllocationClient = _ + private var clock: ManualClock = _ + + // List of known executors. Allows easily mocking which executors are alive without + // having to use mockito APIs directly in each test. + private val knownExecs = mutable.HashSet[String]() + + override def beforeEach(): Unit = { + super.beforeEach() + knownExecs.clear() + clock = new ManualClock() + client = mock(classOf[ExecutorAllocationClient]) + when(client.isExecutorActive(any())).thenAnswer { invocation => + knownExecs.contains(invocation.getArguments()(0).asInstanceOf[String]) + } + monitor = new ExecutorMonitor(conf, client, clock) + } + + test("basic executor timeout") { + knownExecs += "1" + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + assert(monitor.executorCount === 1) + assert(monitor.isExecutorIdle("1")) + assert(monitor.timedOutExecutors(idleDeadline) === Seq("1")) + } + + test("SPARK-4951, SPARK-26927: handle out of order task start events") { + knownExecs ++= Set("1", "2") + + monitor.onTaskStart(SparkListenerTaskStart(1, 1, taskInfo("1", 1))) + assert(monitor.executorCount === 1) + + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + assert(monitor.executorCount === 1) + + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "2", null)) + assert(monitor.executorCount === 2) + + monitor.onExecutorRemoved(SparkListenerExecutorRemoved(clock.getTimeMillis(), "2", null)) + assert(monitor.executorCount === 1) + + knownExecs -= "2" + + monitor.onTaskStart(SparkListenerTaskStart(1, 1, taskInfo("2", 2))) + assert(monitor.executorCount === 1) + } + + test("track tasks running on executor") { + knownExecs += "1" + + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + monitor.onTaskStart(SparkListenerTaskStart(1, 1, taskInfo("1", 1))) + assert(!monitor.isExecutorIdle("1")) + + // Start/end a few tasks and make sure the executor does not go idle. + (2 to 10).foreach { i => + monitor.onTaskStart(SparkListenerTaskStart(i, 1, taskInfo("1", 1))) + assert(!monitor.isExecutorIdle("1")) + + monitor.onTaskEnd(SparkListenerTaskEnd(i, 1, "foo", Success, taskInfo("1", 1), null)) + assert(!monitor.isExecutorIdle("1")) + } + + monitor.onTaskEnd(SparkListenerTaskEnd(1, 1, "foo", Success, taskInfo("1", 1), null)) + assert(monitor.isExecutorIdle("1")) + assert(monitor.timedOutExecutors(clock.getTimeMillis()).isEmpty) + assert(monitor.timedOutExecutors(clock.getTimeMillis() + idleTimeoutMs + 1) === Seq("1")) + } + + test("use appropriate time out depending on whether blocks are stored") { + knownExecs += "1" + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + assert(monitor.isExecutorIdle("1")) + assert(monitor.timedOutExecutors(idleDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(1, 0, "1")) + assert(monitor.isExecutorIdle("1")) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(1, 0, "1", level = StorageLevel.NONE)) + assert(monitor.isExecutorIdle("1")) + assert(monitor.timedOutExecutors(idleDeadline) === Seq("1")) + + monitor.onTaskStart(SparkListenerTaskStart(1, 1, taskInfo("1", 1))) + assert(!monitor.isExecutorIdle("1")) + monitor.onBlockUpdated(rddUpdate(1, 0, "1")) + assert(!monitor.isExecutorIdle("1")) + monitor.onBlockUpdated(rddUpdate(1, 0, "1", level = StorageLevel.NONE)) + assert(!monitor.isExecutorIdle("1")) + } + + test("keeps track of stored blocks for each rdd and split") { + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + + monitor.onBlockUpdated(rddUpdate(1, 0, "1")) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(1, 1, "1")) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(2, 0, "1")) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(1, 1, "1", level = StorageLevel.NONE)) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onUnpersistRDD(SparkListenerUnpersistRDD(1)) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + // Make sure that if we get an unpersist event much later, which moves an executor from having + // cached blocks to no longer having cached blocks, it will time out based on the time it + // originally went idle. + clock.setTime(idleDeadline) + monitor.onUnpersistRDD(SparkListenerUnpersistRDD(2)) + assert(monitor.timedOutExecutors(clock.getTimeMillis()) === Seq("1")) + } + + test("handle timeouts correctly with multiple executors") { + knownExecs ++= Set("1", "2", "3") + + // start exec 1 at 0s (should idle time out at 60s) + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + assert(monitor.isExecutorIdle("1")) + + // start exec 2 at 30s, store a block (should idle time out at 150s) + clock.setTime(TimeUnit.SECONDS.toMillis(30)) + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "2", null)) + monitor.onBlockUpdated(rddUpdate(1, 0, "2")) + assert(monitor.isExecutorIdle("2")) + assert(!monitor.timedOutExecutors(idleDeadline).contains("2")) + + // start exec 3 at 60s (should idle timeout at 120s, exec 1 should time out) + clock.setTime(TimeUnit.SECONDS.toMillis(60)) + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "3", null)) + assert(monitor.timedOutExecutors(clock.getTimeMillis()) === Seq("1")) + + // store block on exec 3 (should now idle time out at 180s) + monitor.onBlockUpdated(rddUpdate(1, 0, "3")) + assert(monitor.isExecutorIdle("3")) + assert(!monitor.timedOutExecutors(idleDeadline).contains("3")) + + // advance to 140s, remove block from exec 3 (time out immediately) + clock.setTime(TimeUnit.SECONDS.toMillis(140)) + monitor.onBlockUpdated(rddUpdate(1, 0, "3", level = StorageLevel.NONE)) + assert(monitor.timedOutExecutors(clock.getTimeMillis()).toSet === Set("1", "3")) + + // advance to 150s, now exec 2 should time out + clock.setTime(TimeUnit.SECONDS.toMillis(150)) + assert(monitor.timedOutExecutors(clock.getTimeMillis()).toSet === Set("1", "2", "3")) + } + + test("SPARK-27677: don't track blocks stored on disk when using shuffle service") { + // First make sure that blocks on disk are counted when no shuffle service is available. + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + monitor.onBlockUpdated(rddUpdate(1, 0, "1", level = StorageLevel.DISK_ONLY)) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + conf.set(SHUFFLE_SERVICE_ENABLED, true).set(SHUFFLE_SERVICE_FETCH_RDD_ENABLED, true) + monitor = new ExecutorMonitor(conf, client, clock) + + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + monitor.onBlockUpdated(rddUpdate(1, 0, "1", level = StorageLevel.MEMORY_ONLY)) + monitor.onBlockUpdated(rddUpdate(1, 1, "1", level = StorageLevel.MEMORY_ONLY)) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(1, 0, "1", level = StorageLevel.DISK_ONLY)) + assert(monitor.timedOutExecutors(idleDeadline).isEmpty) + assert(monitor.timedOutExecutors(storageDeadline) === Seq("1")) + + monitor.onBlockUpdated(rddUpdate(1, 1, "1", level = StorageLevel.DISK_ONLY)) + assert(monitor.timedOutExecutors(idleDeadline) === Seq("1")) + + // Tag the block as being both in memory and on disk, which may happen after it was + // evicted and then restored into memory. Since it's still on disk the executor should + // still be eligible for removal. + monitor.onBlockUpdated(rddUpdate(1, 1, "1", level = StorageLevel.MEMORY_AND_DISK)) + assert(monitor.timedOutExecutors(idleDeadline) === Seq("1")) + } + + test("track executors pending for removal") { + knownExecs ++= Set("1", "2", "3") + + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "1", null)) + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "2", null)) + monitor.onExecutorAdded(SparkListenerExecutorAdded(clock.getTimeMillis(), "3", null)) + clock.setTime(idleDeadline) + assert(monitor.timedOutExecutors().toSet === Set("1", "2", "3")) + assert(monitor.pendingRemovalCount === 0) + + // Notify that only a subset of executors was killed, to mimic the case where the scheduler + // refuses to kill an executor that is busy for whatever reason the monitor hasn't detected yet. + monitor.executorsKilled(Seq("1")) + assert(monitor.timedOutExecutors().toSet === Set("2", "3")) + assert(monitor.pendingRemovalCount === 1) + + // Check the timed out executors again so that we're sure they're still timed out when no + // events happen. This ensures that the monitor doesn't lose track of them. + assert(monitor.timedOutExecutors().toSet === Set("2", "3")) + + monitor.onTaskStart(SparkListenerTaskStart(1, 1, taskInfo("2", 1))) + assert(monitor.timedOutExecutors().toSet === Set("3")) + + monitor.executorsKilled(Seq("3")) + assert(monitor.pendingRemovalCount === 2) + + monitor.onTaskEnd(SparkListenerTaskEnd(1, 1, "foo", Success, taskInfo("2", 1), null)) + assert(monitor.timedOutExecutors().isEmpty) + clock.advance(idleDeadline) + assert(monitor.timedOutExecutors().toSet === Set("2")) + } + + private def idleDeadline: Long = clock.getTimeMillis() + idleTimeoutMs + 1 + private def storageDeadline: Long = clock.getTimeMillis() + storageTimeoutMs + 1 + + private def taskInfo( + execId: String, + id: Int, + speculative: Boolean = false, + duration: Long = -1L): TaskInfo = { + val start = if (duration > 0) clock.getTimeMillis() - duration else clock.getTimeMillis() + val task = new TaskInfo(id, id, 1, start, execId, "foo.example.com", + TaskLocality.PROCESS_LOCAL, speculative) + if (duration > 0) { + task.markFinished(TaskState.FINISHED, math.max(1, clock.getTimeMillis())) + } + task + } + + private def rddUpdate( + rddId: Int, + splitIndex: Int, + execId: String, + level: StorageLevel = StorageLevel.MEMORY_ONLY): SparkListenerBlockUpdated = { + SparkListenerBlockUpdated( + BlockUpdatedInfo(BlockManagerId(execId, "1.example.com", 42), + RDDBlockId(rddId, splitIndex), level, 1L, 0L)) + } + +} diff --git a/core/src/test/scala/org/apache/spark/storage/BlockManagerInfoSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockManagerInfoSuite.scala index 8df1232..01e3d6a 100644 --- a/core/src/test/scala/org/apache/spark/storage/BlockManagerInfoSuite.scala +++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerInfoSuite.scala @@ -47,7 +47,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { broadcastId, StorageLevel.MEMORY_AND_DISK, memSize = 200, diskSize = 100) assert(bmInfo.blocks.asScala === Map(broadcastId -> BlockStatus(StorageLevel.MEMORY_AND_DISK, 0, 100))) - assert(bmInfo.exclusiveCachedBlocks.isEmpty) assert(bmInfo.remainingMem === 29800) } @@ -56,7 +55,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { bmInfo.updateBlockInfo(rddId, StorageLevel.MEMORY_ONLY, memSize = 200, diskSize = 0) assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.MEMORY_ONLY, 200, 0))) - assert(bmInfo.exclusiveCachedBlocks === Set(rddId)) assert(bmInfo.remainingMem === 29800) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.isEmpty) @@ -70,8 +68,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { bmInfo.updateBlockInfo(rddId, StorageLevel.MEMORY_AND_DISK, memSize = 200, diskSize = 400) assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.MEMORY_AND_DISK, 0, 400))) - val exclusiveCachedBlocksForOneMemoryOnly = if (svcEnabled) Set() else Set(rddId) - assert(bmInfo.exclusiveCachedBlocks === exclusiveCachedBlocksForOneMemoryOnly) assert(bmInfo.remainingMem === 29800) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.asScala === @@ -85,7 +81,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.DISK_ONLY, 0, 200))) val exclusiveCachedBlocksForOneMemoryOnly = if (svcEnabled) Set() else Set(rddId) - assert(bmInfo.exclusiveCachedBlocks === exclusiveCachedBlocksForOneMemoryOnly) assert(bmInfo.remainingMem === 30000) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.asScala === @@ -99,7 +94,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { val rddId: BlockId = RDDBlockId(0, 0) bmInfo.updateBlockInfo(rddId, StorageLevel.MEMORY_ONLY, memSize = 200, 0) assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.MEMORY_ONLY, 200, 0))) - assert(bmInfo.exclusiveCachedBlocks === Set(rddId)) assert(bmInfo.remainingMem === 29800) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.isEmpty) @@ -107,8 +101,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { bmInfo.updateBlockInfo(rddId, StorageLevel.DISK_ONLY, memSize = 0, diskSize = 200) assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.DISK_ONLY, 0, 200))) - val exclusiveCachedBlocksForNoMemoryOnly = if (svcEnabled) Set() else Set(rddId) - assert(bmInfo.exclusiveCachedBlocks === exclusiveCachedBlocksForNoMemoryOnly) assert(bmInfo.remainingMem === 30000) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.asScala === @@ -120,8 +112,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { val rddId: BlockId = RDDBlockId(0, 0) bmInfo.updateBlockInfo(rddId, StorageLevel.DISK_ONLY, memSize = 0, diskSize = 200) assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.DISK_ONLY, 0, 200))) - val exclusiveCachedBlocksForOneMemoryOnly = if (svcEnabled) Set() else Set(rddId) - assert(bmInfo.exclusiveCachedBlocks === exclusiveCachedBlocksForOneMemoryOnly) assert(bmInfo.remainingMem === 30000) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.asScala === @@ -130,7 +120,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { bmInfo.updateBlockInfo(rddId, StorageLevel.NONE, memSize = 0, diskSize = 200) assert(bmInfo.blocks.isEmpty) - assert(bmInfo.exclusiveCachedBlocks.isEmpty) assert(bmInfo.remainingMem === 30000) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.isEmpty) @@ -141,8 +130,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { val rddId: BlockId = RDDBlockId(0, 0) bmInfo.updateBlockInfo(rddId, StorageLevel.DISK_ONLY, memSize = 0, diskSize = 200) assert(bmInfo.blocks.asScala === Map(rddId -> BlockStatus(StorageLevel.DISK_ONLY, 0, 200))) - val exclusiveCachedBlocksForOneMemoryOnly = if (svcEnabled) Set() else Set(rddId) - assert(bmInfo.exclusiveCachedBlocks === exclusiveCachedBlocksForOneMemoryOnly) assert(bmInfo.remainingMem === 30000) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.asScala === @@ -151,7 +138,6 @@ class BlockManagerInfoSuite extends SparkFunSuite { bmInfo.removeBlock(rddId) assert(bmInfo.blocks.asScala.isEmpty) - assert(bmInfo.exclusiveCachedBlocks.isEmpty) assert(bmInfo.remainingMem === 30000) if (svcEnabled) { assert(bmInfo.externalShuffleServiceBlockStatus.get.isEmpty) diff --git a/streaming/src/test/scala/org/apache/spark/streaming/scheduler/ExecutorAllocationManagerSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/scheduler/ExecutorAllocationManagerSuite.scala index d072b99..a8b0055 100644 --- a/streaming/src/test/scala/org/apache/spark/streaming/scheduler/ExecutorAllocationManagerSuite.scala +++ b/streaming/src/test/scala/org/apache/spark/streaming/scheduler/ExecutorAllocationManagerSuite.scala @@ -388,7 +388,7 @@ class ExecutorAllocationManagerSuite extends SparkFunSuite } private def withStreamingContext(conf: SparkConf)(body: StreamingContext => Unit): Unit = { - conf.setMaster("myDummyLocalExternalClusterManager") + conf.setMaster("local-cluster[1,1,1024]") .setAppName(this.getClass.getSimpleName) .set("spark.streaming.dynamicAllocation.testing", "true") // to test dynamic allocation --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org