srowen commented on a change in pull request #22015:
[SPARK-20286][SPARK-24786][Core][DynamicAllocation] Release executors on
unpersisting RDD
URL: https://github.com/apache/spark/pull/22015#discussion_r258183329
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -600,28 +600,48 @@ private[spark] class ExecutorAllocationManager(
* 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. Note that
this
- // does not include broadcast blocks.
- val hasCachedBlocks = blockManagerMaster.hasCachedBlocks(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
+ onExecutorIdle(Seq(executorId))
+ }
+
+ private def onExecutorIdle(executorIds: Seq[String], onUnpersist: Boolean =
false): Unit =
+ synchronized {
+ executorIds.foreach(executorId => {
+ if (executorIds.contains(executorId)) {
Review comment:
These two lines seem redundant; by definition the seocnd line is true
because of the new one before it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
