HyukjinKwon commented on code in PR #59196:
URL: https://github.com/apache/spark/pull/59196#discussion_r4265794228
##########
common/utils/src/main/scala/org/apache/spark/util/MavenUtils.scala:
##########
@@ -62,32 +63,102 @@ private[spark] object MavenUtils extends Logging {
}
}
- private def acquireIvyLock(isCancelled: () => Boolean): Unit = {
- def checkCancelled(): Unit = {
- if (isCancelled() || Thread.currentThread().isInterrupted) {
- throw new CancellationException("Maven dependency resolution was
cancelled")
- }
+ private class IvyResolutionCancellation {
+ private val requested = new AtomicBoolean(false)
+
+ def isCancelled: Boolean = requested.get()
+
+ def cancel(resolutionThread: Thread): Unit = {
+ requested.set(true)
+ resolutionThread.interrupt()
+ }
+ }
+
+ private def checkCancelled(isCancelled: () => Boolean): Unit = {
+ if (isCancelled() || Thread.currentThread().isInterrupted) {
+ throw new CancellationException("Maven dependency resolution was
cancelled")
}
+ }
- checkCancelled()
+ private def acquireIvyLock(isCancelled: () => Boolean): Unit = {
+ checkCancelled(isCancelled)
try {
while (!ivyLock.tryLock(ivyLockPollIntervalMs, TimeUnit.MILLISECONDS)) {
- checkCancelled()
+ checkCancelled(isCancelled)
}
} catch {
case _: InterruptedException =>
Thread.currentThread().interrupt()
throw new CancellationException("Maven dependency resolution was
cancelled")
}
try {
- checkCancelled()
+ checkCancelled(isCancelled)
} catch {
case e: Throwable =>
ivyLock.unlock()
throw e
}
}
+ private def runWithCancellation[T](isCancelled: () => Boolean)(
+ body: IvyResolutionCancellation => T): T = {
+ checkCancelled(isCancelled)
+ val cancellation = new IvyResolutionCancellation
+ val task = new FutureTask[T](new Callable[T] {
+ override def call(): T = body(cancellation)
+ })
+ val resolutionThread = new Thread(task, "spark-ivy-resolution")
+ resolutionThread.setDaemon(true)
+ resolutionThread.start()
+
+ def cancelAndWait(): Unit = {
+ var wasInterrupted = Thread.interrupted()
+ try {
+ cancellation.cancel(resolutionThread)
+ while (resolutionThread.isAlive) {
+ try {
+ resolutionThread.join()
+ } catch {
+ case _: InterruptedException => wasInterrupted = true
Review Comment:
**Nit (P3):** `cancelAndWait` manages the caller's interrupt status in three
places:
- it clears the status with `Thread.interrupted()` (line 115);
- it records an interrupt that arrives during the join (this line);
- it restores the status in its `finally` block (lines 126-127).
That `finally` block is the only restore on the `CancellationException` arm
(lines 151-153). No current test fails if the join-loop handling or the
`finally` restore regresses.
'cancellable Maven resolution restores caller interruption after joining the
worker' interrupts the caller while it is blocked inside `isCancelled`. The
resulting `InterruptedException` clears the flag before `cancelAndWait` runs,
so `Thread.interrupted()` returns false and the join is never interrupted. The
`callerInterrupted` assertion is then satisfied by the separate re-interrupt at
line 149 alone. The other cancellation tests never interrupt the caller.
Both untested paths are reachable in production:
- The Connect predicate includes `Thread.currentThread().isInterrupted`, so
`checkCancelled` can raise `CancellationException` with the flag still set.
- A caller can be interrupted again while the worker sits in a socket read
until the read timeout.
If this catch or the `finally` restore were removed, the caller could return
while the worker still holds `ivyLock` and the redirected `System.out`, or it
could lose its interrupt status. The suite would still pass in both cases.
The existing blocking-resolver caller-interrupt test could be extended to
cover both paths:
1. Interrupt the caller a second time after `resolverInterrupted` fires.
2. Add a variant whose `isCancelled` callback calls
`Thread.currentThread().interrupt()` and returns false, so cancellation enters
through the still-set flag.
**Verification:**
- **Regression:** If the join loop lets an InterruptedException escape, a
test that interrupts the caller during the join fails because the caller
returns before the worker terminates.
- **Regression:** If cancelAndWait's finally interrupt restore is removed, a
test that enters cancellation through the caller's interrupt flag fails because
the caller's interrupt status is lost.
- **Behavior:** Against the current implementation, the new coverage passes
and leaves no caller or worker thread running.
--
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]
