pkotikalapudi commented on code in PR #42352:
URL: https://github.com/apache/spark/pull/42352#discussion_r1577078911
##########
core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala:
##########
@@ -340,6 +385,45 @@ private[spark] class ExecutorAllocationManager(
}
}
+ /**
+ * Maximum number of executors to be removed per dra evaluation.
+ *
+ * This function limits the number of idle executors to be removed if the
+ * `streamingDRAFeatureEnabled` flag is enabled, otherwise it returns all
the idle executors.
+ * The number of executors to be removed is based on
+ * the de-allocation ratio (`executorDeallocationRatio`) and timeout
(`removeTime`).
+ * It helps in removing only few executors per evaluation cycle and helps in
gradual removal of
+ * executors across micro-batches.
+ */
+ private def maxExecutorDeallocationsPerEvaluation(
+ timedOutExecs :
Seq[(String, Int)]
+ ): Seq[(String, Int)] = {
+ if (!streamingDRAFeatureEnabled) {
+ timedOutExecs
+ }
+ else {
+ val currentTime = clock.nanoTime()
+
+ if (removeTime == NOT_SET) {
Review Comment:
Good catch. I will update it to `Seq.empty[(String, Int)]`. Every evaluation
happens at a fixed rate of few hundred milliseconds. So all those executors
will be evaluated in the next round
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
