sarutak commented on code in PR #59126:
URL: https://github.com/apache/spark/pull/59126#discussion_r4129995044
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala:
##########
@@ -261,6 +261,8 @@ private[spark] object Config extends Logging {
"off-heap memory. " +
"TOTAL_SHUFFLE_WRITE policy chooses an executor with the biggest total
shuffle write. " +
"DISK_USED policy chooses an executor with the biggest used disk size.
" +
+ "ACTIVE_TASKS policy chooses an executor with the smallest number of
active tasks. " +
Review Comment:
The doc says the policy "chooses an executor with the smallest number of
active tasks", but `choose()` first filters candidates by `totalTasks >=
minTasks` (`MINIMUM_TASKS_PER_EXECUTOR_BEFORE_ROLLING`). So an executor with
`activeTasks == 0` but `totalTasks < minTasks` is excluded, and the chosen
executor may not have the globally smallest `activeTasks`.
This is the existing behavior shared by all policies (covered by the test
`"SPARK-37806: All policy should ignore executor if totalTasks < minTasks"` in
`ExecutorRollPluginSuite`), so no code change is needed. But since the
`ACTIVE_TASKS` description is phrased as an absolute, a short note that
executors protected by `minTasksPerExecutorBeforeRolling` are not candidates
would avoid confusion.
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorRollPlugin.scala:
##########
@@ -135,6 +135,8 @@ class ExecutorRollDriverPlugin extends DriverPlugin with
Logging {
listWithoutDriver.sortBy(_.totalShuffleWrite).reverse
case ExecutorRollPolicy.DISK_USED =>
listWithoutDriver.sortBy(_.diskUsed).reverse
+ case ExecutorRollPolicy.ACTIVE_TASKS =>
Review Comment:
A one-line comment would help future maintainers understand that the
tie-breaker is intentional rather than accidental.
```scala
case ExecutorRollPolicy.ACTIVE_TASKS =>
// Active-task counts tie very frequently (all busy or all idle), so break
// ties by the oldest add-time to keep the choice deterministic.
listWithoutDriver.sortBy(e => (e.activeTasks, e.addTime.getTime))
```
--
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]
