jerrypeng commented on code in PR #42352:
URL: https://github.com/apache/spark/pull/42352#discussion_r1566270204
##########
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:
Why do we remove all the timedOutExecs when removeTime is not set? Doesn't
this mean for the first evaluation of this function when removeTime is not set,
we will remove all timeOutExecs and NOT adhere to the configured
"executorDeallocationRatio"?
--
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]
