HyukjinKwon commented on code in PR #59196:
URL: https://github.com/apache/spark/pull/59196#discussion_r4256497046
##########
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
+ }
+ }
+ } finally {
+ if (wasInterrupted) {
+ Thread.currentThread().interrupt()
+ }
+ }
+ }
+
+ try {
+ var result: T = null.asInstanceOf[T]
+ var completed = false
+ while (!completed) {
+ checkCancelled(isCancelled)
+ try {
+ result = task.get(ivyLockPollIntervalMs, TimeUnit.MILLISECONDS)
+ completed = true
+ } catch {
+ case _: TimeoutException =>
+ }
+ }
+ checkCancelled(isCancelled)
+ result
+ } catch {
+ case _: InterruptedException =>
Review Comment:
**Nit (P3):** `runWithCancellation` adds two exits besides the cancellation
flag, and neither has a test that would fail if it regressed. The first is the
worker-failure unwrapping (`case e: ExecutionException => throw e.getCause`,
lines 154-155). The second is this caller-interrupt branch, which cancels and
joins the worker, restores the caller's interrupt flag, and throws
`CancellationException`. Every new runtime-resolver test signals cancellation
only through `isCancelled` and asserts only `CancellationException`. The
existing Ivy failure tests call the synchronous `resolveMavenCoordinates`,
which never enters this routing, and the new AddArtifactsHandlerSuite failure
test is rejected by coordinate validation before MavenUtils runs.
Two regressions would pass the whole suite. If the `ExecutionException` arm
were dropped or reordered, classic `addArtifact(ivy://...)` and Connect callers
would receive a checked `ExecutionException` wrapper instead of Ivy's
unresolved-dependency `RuntimeException`. If the interrupt restore were
dropped, or the method returned before the join, an interrupted classic caller
would lose its interrupt status or return while the worker still holds
`ivyLock`. The current routing is correct; only regression coverage is missing.
**Verification:**
- **Regression:** Removing or reordering the ExecutionException unwrapping
makes the failure-propagation test fail.
- **Regression:** Dropping the interrupt-status restoration, or returning
before the worker is joined, makes the caller-interrupt test fail.
- **Behavior:** Both tests pass against the current implementation and leave
no executor or worker thread behind.
##########
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()
Review Comment:
**Non-blocking (P2):** `IvyResolutionCancellation.cancel` interrupts the
worker in whatever phase it is in, including during `ivy.retrieve` /
`noCacheIvy.retrieve`. The new `checkCancelled` calls only bracket retrieve, so
they cannot stop an in-flight copy from being cut short. In Ivy 2.5.3,
`RetrieveEngine` copies each cached jar straight into `<ivyPath>/jars` with
`FileUtil.copy(File, File, null, true)`. When the thread is interrupted, the
copy loop throws `IOException("transfer interrupted")`, which skips both the
size-check-and-delete and the `setLastModified(src)` step. The partial jar
therefore stays in place with a fresh mtime. MavenUtils keeps the default
`"newer"` overwrite mode, under which `RetrieveEngine.upToDate` treats any
target at least as new as the cache source as current.
For runtime resolution, `ArtifactManager` uses a per-session
`<artifactPath>/.ivy-cache`. After a Connect client cancels, or its gRPC
deadline expires, while a jar is being retrieved, the next resolution of that
coordinate in the same session skips the copy. `resolveDependencyPaths` then
returns the truncated jar, and the AddArtifacts handler registers it on the
session classpath. Later class loading fails with confusing `ZipException` /
`ClassNotFoundException` errors until the session is recreated. A caller
interrupt during the synchronous retrieve could already do this on master. This
PR makes it reachable from ordinary flag-driven cancellation and deadline
expiry, which previously always let the copy finish. No test cancels during
retrieve. The synchronous `resolveMavenCoordinates` path is not affected by
this change.
**Recommended change:** On the cancellable path (cancellation is defined),
configure the retrieve options with RetrieveOptions.OVERWRITEMODE_DIFFERENT, so
a retrieved target counts as up to date only when its mtime equals the cached
artifact's. Leave the synchronous path's retrieve options unchanged. Add a
MavenUtilsSuite regression test that leaves a truncated, newer-mtime jar in the
runtime resolver's packages directory and checks that a following runtime
resolution returns a complete copy.
**Why this works:** Every complete Ivy retrieve copy stamps the destination
with the source's lastModified. A copy cut short by an interrupt, or by any
other mid-copy IOException, keeps the current time. With mtime-equality as the
up-to-date test, a partial file always fails the check, and the next retrieve
overwrites it with a full copy. Complete files from earlier resolutions keep
matching and are not recopied, and the cancellation and interrupt behavior is
unchanged.
**Scope:** Tighten the runtime (cancellable) resolution's retrieve
up-to-date policy so partial jars left by a cancelled retrieve are recopied
rather than reused, and cover it with a regression test.
**Compatibility:** Cancellation semantics, interrupt delivery, and the
synchronous resolution path are unchanged.
**Risks:** If a filesystem does not preserve millisecond lastModified across
setLastModified, complete jars may be recopied on each runtime resolution. That
costs extra local copying but is not incorrect.
**Constraints:** Must not change the synchronous resolveMavenCoordinates
retrieve behavior used by spark-submit --packages, SparkContext.addJar, and
Hive loading. Must keep the cooperative cancel-then-join lifecycle, the
cancellation checkpoints, and the existing URL handler and transport unchanged.
**Success:** After a cancelled runtime resolution leaves a partial jar in
the per-session packages directory, the next successful runtime resolution of
that artifact returns a complete jar identical to the cached artifact. Runtime
resolutions whose retrieved jars are already complete do not fail, and still
return the same paths. Synchronous resolveMavenCoordinates keeps Ivy's default
retrieve overwrite behavior.
**Verification:**
- **Regression:** A runtime resolution whose packages directory already
holds a truncated copy of the artifact, with an mtime newer than the cached
artifact, returns a jar whose size and content match the cached artifact.
- **Behavior:** Repeated successful runtime resolutions of the same
coordinate in one ivy path still succeed and return the same jar paths.
- **Compatibility:** Existing synchronous resolution and runtime
cancellation tests keep passing, confirming that only the cancellable path's
retrieve policy changed.
--
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]
