sunchao commented on code in PR #59756:
URL: https://github.com/apache/spark/pull/59756#discussion_r4396982374
##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala:
##########
@@ -654,7 +810,31 @@ class ExecutorPodsAllocator(
}
}
+ override def stopAllocation(): Unit = synchronized {
+ if (!podAllocationParallelEnabled || stopped) return
+ stopped = true
+ podCreationThreadPool.foreach { pool =>
+ // Do not interrupt an in-flight create: Fabric8 stops waiting for the
response on interrupt,
+ // but the API server may still create the resource. Let workers observe
the result and clean
+ // up before the backend deletes PVCs and closes the client. Queued
tasks need not run.
+ pool.shutdown()
+ val queuedTasks = new ArrayList[Runnable]()
+ pool.getQueue.drainTo(queuedTasks)
Review Comment:
[P2] Wake idle workers after draining the shutdown queue
There is a race between `shutdown()` and this direct queue drain when an
active creation finishes: `ThreadPoolExecutor.getTask()` can observe SHUTDOWN
with a nonempty queue, then call `LinkedBlockingQueue.take()` after `drainTo`
has emptied it. That worker remains blocked with no active requests or queued
work. `awaitTermination` then spends the entire configured timeout (20s by
default), logs an in-flight-request warning, and leaves a daemon worker alive;
repeated `stopAllocation()` returns early and cannot recover it.
I reproduced this on JDK 17 and 21 with the same fixed-pool
shutdown/drain/cancel sequence, both naturally and with the interleaving
forced. The stuck state is `active=0, queued=0, terminated=false`; signaling
idle workers with another `shutdown()` makes it terminate. Please use
termination-aware queue removal or signal idle workers again after draining,
while preserving the requirement not to interrupt active API calls. Please also
cover a request finishing *during* the drain: the current tests release the
response after draining has settled and miss this race.
##########
docs/running-on-kubernetes.md:
##########
@@ -1006,6 +1006,47 @@ See the [configuration page](configuration.html) for
information on Spark config
</td>
<td>4.2.0</td>
</tr>
+<tr>
+ <td><code>spark.kubernetes.allocation.batch.parallel.enabled</code></td>
+ <td><code>false</code></td>
+ <td>
+ Create executor pods concurrently within each allocation batch. Pod
specifications and PVC reuse
Review Comment:
Nonblocking documentation clarification: could we state that these three
settings apply to `spark.kubernetes.allocation.pods.allocator=direct`? Only
`ExecutorPodsAllocator` reads them; the Deployment and StatefulSet allocators
do not use the parallel switch, concurrency, or shutdown timeout. The current
wording can lead users of those allocators to expect these options to take
effect. It would help to add the same scope to the configuration descriptions
in `Config.scala`.
--
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]
