(spark) branch master updated: [SPARK-53394][CORE] UninterruptibleLock.isInterruptible should avoid duplicated interrupt

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/master by this push:
     new 78871d7de88b [SPARK-53394][CORE] UninterruptibleLock.isInterruptible should avoid duplicated interrupt
78871d7de88b is described below

commit 78871d7de88b30729f543f1a0fef0b845d04a37e
Author: Yi Wu <yi...@databricks.com>
AuthorDate: Fri Aug 29 10:46:09 2025 +0800

    [SPARK-53394][CORE] UninterruptibleLock.isInterruptible should avoid duplicated interrupt
    
    ### What changes were proposed in this pull request?
    
    This PR fixes `UninterruptibleLock.isInterruptible` to avoid duplicated interruption when the thread is already interrupted.
    
    ### Why are the changes needed?
    
    The "uninterruptible" semantic of `UninterruptibleThread`is broken (i.e., `UninterruptibleThread` is interruptible even if it's under `runUninterruptibly`) after the fix https://github.com/apache/spark/pull/50594. The probelm is that the state of
    `shouldInterruptThread` becomes unsafe when there are multiple interrupts concurrently.
    
    For example, thread A could interrupt UninterruptibleThread ut first before UninterruptibleThread enters `runUninterruptibly`. Right after that, another thread B starts to invoke ut.interrupt() and pass through `uninterruptibleLock.isInterruptible` (becasue at this point, `shouldInterruptThread = uninterruptible = false`). Before thread B invokes `super.interrupt()`, UninterruptibleThread ut enters `runUninterruptibly` and pass through `uninterruptibleLock.getAndSetUninterruptible` an [...]
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    Manually tested. The issue can be easily reproduced if we run `UninterruptibleThreadSuite.stress test` for 100 times in a row:
    ```
    [info]   true did not equal false (UninterruptibleThreadSuite.scala:208)
    [info]   org.scalatest.exceptions.TestFailedException:
    [info]   at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
    [info]   at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
    [info]   at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
    [info]   at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
    [info]   at org.apache.spark.util.UninterruptibleThreadSuite.$anonfun$new$7(UninterruptibleThreadSuite.scala:208)
    ...
    ```
    And the issue is gone after the fix.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    No
    
    Closes #52139 from Ngone51/fix-uninterruptiable.
    
    Authored-by: Yi Wu <yi...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../main/scala/org/apache/spark/util/UninterruptibleThread.scala   | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala b/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala
index 8fba5ed944c6..3d504856f566 100644
--- a/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala
+++ b/core/src/main/scala/org/apache/spark/util/UninterruptibleThread.scala
@@ -107,8 +107,11 @@ private[spark] class UninterruptibleThread(
       // super.interrupt() is called. In this case to prevent runUninterruptibly() from being
       // interrupted, we use awaitInterruptThread flag. We need to set it only if
       // runUninterruptibly() is not yet set uninterruptible to true (!shouldInterruptThread) and
-      // there is no other threads that called interrupt (awaitInterruptThread is already true)
-      if (!shouldInterruptThread && !awaitInterruptThread) {
+      // there is no other threads that called interrupt (awaitInterruptThread is already true or
+      // isInterrupted is true. (SPARK-53394) Otherwise, the state of shouldInterruptThread would
+      // become inconsistent between isInterruptible() and isInterruptPending(), leading to
+      // UninterruptibleThread be interruptible under runUninterruptibly.)
+      if (!shouldInterruptThread && !awaitInterruptThread && !isInterrupted) {
         awaitInterruptThread = true
         true
       } else {


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org