HyukjinKwon commented on code in PR #59196:
URL: https://github.com/apache/spark/pull/59196#discussion_r4162068454
##########
common/utils/src/test/scala/org/apache/spark/util/MavenUtilsSuite.scala:
##########
@@ -283,6 +295,68 @@ class MavenUtilsSuite
}
}
+ test("runtime dependency resolver cancels active Ivy resolution") {
+ val executor = Executors.newFixedThreadPool(2)
+ implicit val executionContext: ExecutionContext =
+ ExecutionContext.fromExecutorService(executor)
+ val server = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))
+ val requestAccepted = new CountDownLatch(1)
+ val releaseRequest = new CountDownLatch(1)
+ val serverFuture = Future {
+ val socket = server.accept()
+ try {
+ requestAccepted.countDown()
+ releaseRequest.await()
+ } finally {
+ socket.close()
+ }
+ }
+ try {
+ val settings = Paths.get(tempIvyPath, "cancellable-ivysettings.xml")
+ Files.writeString(
+ settings,
+ s"""<ivysettings>
+ | <settings defaultResolver="blocked"/>
+ | <resolvers>
+ | <ibiblio name="blocked" m2compatible="true"
+ | root="http://127.0.0.1:${server.getLocalPort}"/>
+ | </resolvers>
+ |</ivysettings>""".stripMargin)
+ val cancelled = new AtomicBoolean(false)
+ val resolver = new RuntimeDependencyResolver(
+ ivySettingsPath = Some(settings.toString),
+ ivyPath = Some(Paths.get(tempIvyPath, "cancelled").toString))
+ val resolution = Future {
+ resolver.resolve(
+ URI.create("ivy://my.runtime.active:mylib:0.1"),
+ resolverTimeoutMs,
+ resolverTimeoutMs,
+ isCancelled = () => cancelled.get())
+ }
+
+ assert(requestAccepted.await(5, TimeUnit.SECONDS))
+ cancelled.set(true)
+ intercept[CancellationException] {
+ SparkThreadUtils.awaitResultNoSparkExceptionConversion(resolution,
5.seconds)
+ }
+
+ val next = MavenCoordinate("my.runtime.after.cancel", "mylib", "0.1")
+ IvyTestUtils.withRepository(next, None, None) { repo =>
Review Comment:
**Nit (P3):** This follow-up resolution proves that `ivyLock` was released,
but it uses a local file repository, so it never dispatches through the
http/https `HttpClientHandler`. If the `finally` block stopped restoring the
previous `URLHandlerRegistry` default, the closed per-resolution dispatcher
would stay installed JVM-wide and every later HTTP(S) resolution, including
`--packages`, would fail, yet this test and the others would still pass.
Capturing `URLHandlerRegistry.getDefault` before the resolution and asserting
that the same instance is back after both the cancelled and a successful
cancellable resolution would close this gap.
**Verification:**
- **Regression:** After a cancelled resolution and after a successful
cancellable resolution, the global Ivy URL handler is identical to the one
installed before the resolution started.
##########
common/utils/src/main/scala/org/apache/spark/util/MavenUtils.scala:
##########
@@ -53,32 +56,132 @@ private[spark] object MavenUtils extends Logging {
private val ivyLock = new ReentrantLock()
private val ivyLockPollIntervalMs = 100L
- 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)
+ private val activeIvy = new AtomicReference[Ivy]()
+ private val httpHandler = new AtomicReference[HttpClientHandler]()
+
+ def isCancelled: Boolean = requested.get()
+
+ def setActiveIvy(ivy: Ivy): Unit = activeIvy.set(ivy)
+
+ def installUrlHandler(): () => Unit = {
+ val previousHandler = URLHandlerRegistry.getDefault
+ val handler = new HttpClientHandler
+ httpHandler.set(handler)
+ val dispatcher = new URLHandlerDispatcher
+ dispatcher.setDefault(previousHandler)
+ dispatcher.setDownloader("http", handler)
+ dispatcher.setDownloader("https", handler)
+ URLHandlerRegistry.setDefault(dispatcher)
+ () => URLHandlerRegistry.setDefault(previousHandler)
+ }
+
+ def closeHttpHandler(): Unit = {
+ Option(httpHandler.getAndSet(null)).foreach { handler =>
+ try {
+ handler.close()
+ } catch {
+ case NonFatal(e) => logWarning("Failed to close Ivy HTTP handler", e)
+ }
+ }
+ }
+
+ def cancel(resolutionThread: Thread): Unit = {
+ requested.set(true)
+ closeHttpHandler()
+ Option(activeIvy.get()) match {
+ case Some(ivy) => ivy.interrupt(resolutionThread)
Review Comment:
**Non-blocking (P2):** `Ivy.interrupt(Thread)` in Ivy 2.5.3 calls
`Thread.interrupt()`, joins the thread for `IvySettings.getInterruptTimeout()`
(a hard-coded 2000 ms), and then calls `Thread.stop()` if the thread is still
alive. Closing the HttpClientHandler handles the blocked-HTTP-read case. If the
worker is still busy after 2 s, for example in a DNS lookup, a TCP connect
within the connect timeout, a file or non-HTTP resolver, or local
cache/retrieve I/O, then:
- on JDK 21+, `Thread.stop()` throws `UnsupportedOperationException`. It
escapes `cancel()` and `cancelAndWait()` before the join loop runs, so the
caller gets that exception instead of `CancellationException` and returns while
the worker still holds `ivyLock` and the swapped global `URLHandlerRegistry`
default;
- on JDK 17, the worker is stopped with an asynchronous `ThreadDeath` in the
middle of Ivy resolution or cache writes, or even inside the `finally` that
restores the handler and `System.out`.
The new test only covers the case where closing the HTTP client unblocks the
worker in time, so it does not exercise this path.
**Recommended change:** In IvyResolutionCancellation.cancel, stop calling
Ivy.interrupt(resolutionThread). Set the requested flag, close the
per-resolution HttpClientHandler, and call resolutionThread.interrupt()
directly. Then let cancelAndWait's existing join loop wait for the worker.
Optionally re-check cancellation at Spark-owned points between Ivy phases (for
example around the retrieve step) so non-network work stops at the next
checkpoint.
**Why this works:** Without Ivy.interrupt, Thread.stop is never invoked.
Thread interrupt plus closing the HTTP client unblocks network I/O, the
requested flag makes the next checkCancelled throw, and cancelAndWait always
reaches its join. The worker's finally block therefore runs normally, releasing
ivyLock and restoring the URL handler and System.out before the caller throws
CancellationException.
**Scope:** Make cancellation of the Ivy worker thread cooperative and remove
Ivy's Thread.stop fallback.
**Compatibility:** Prompt cancellation of a resolution blocked on an HTTP
read, and the unchanged non-cancellable resolveMavenCoordinates path.
**Risks:** Without Ivy's checkInterrupted flag, a long non-network
resolution may run to completion after cancellation, delaying the caller's
CancellationException.
**Constraints:** Must not call Thread.stop directly or indirectly. Must keep
the interrupt status handling and the ExecutionException unwrapping in
runWithCancellation.
**Success:** The cancellation path never invokes Thread.stop on any JDK.
When cancellation is requested, the caller receives CancellationException only
after the worker thread has terminated, ivyLock has been released, and the
previous URLHandlerRegistry default has been restored.
**Verification:**
- **Regression:** Cancel a resolution whose worker cannot be unblocked by
closing the HTTP client within 2 s (for example, a blocking step that ignores
interrupts for longer than the Ivy interrupt timeout). Verify that the caller
receives CancellationException rather than UnsupportedOperationException and
that the worker has terminated when it does.
- **Behavior:** The existing blocked-HTTP-read cancellation scenario still
raises CancellationException promptly and leaves the lock free for a following
resolution.
--
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]
