(spark) branch master updated: [SPARK-58245][CORE] Fix AsyncEventQueue Shutdown Hang When Poison Pill Enqueue Is Interrupted

[we...@apache.org]

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

cloud-fan 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 ce39ba8e7e02 [SPARK-58245][CORE] Fix AsyncEventQueue Shutdown Hang When Poison Pill Enqueue Is Interrupted
ce39ba8e7e02 is described below

commit ce39ba8e7e023123dcb62548443e268c946459f3
Author: Zerui Bao <ze...@databricks.com>
AuthorDate: Wed Jul 22 19:42:56 2026 +0800

    [SPARK-58245][CORE] Fix AsyncEventQueue Shutdown Hang When Poison Pill Enqueue Is Interrupted
    
    ### What changes were proposed in this pull request?
    This PR makes `AsyncEventQueue` shutdown robust to thread interruption. In `AsyncEventQueue.stop()`, two blocking operations are made uninterruptible:
    
    - Enqueuing the POISON_PILL now uses `Uninterruptibles.putUninterruptibly(eventQueue, POISON_PILL)` instead of eventQueue.put(POISON_PILL).
    - Joining the dispatch thread now uses Uninterruptibles.joinUninterruptibly(...), preserving the existing `spark.scheduler.listenerbus.exitTimeout` semantics (wait indefinitely when the timeout is 0, otherwise wait up to the configured timeout).
    Both operations restore the caller's interrupt status before returning, so interruption is not swallowed.
    
    ### Why are the changes needed?
    `AsyncEventQueue.stop()` can be invoked from a thread that has already been interrupted. For example, a Structured Streaming execution thread being torn down. In that case `eventQueue.put(POISON_PILL)` throws InterruptedException before the poison pill is enqueued. Because stopped has already been flipped to true, no further poison pill is ever enqueued, and the dispatch thread (spark-listener-group-*) blocks forever on `eventQueue.take()`, leaking the thread and preventing a clean sh [...]
    
    ### Does this PR introduce _any_ user-facing change?
    No
    
    ### How was this patch tested?
    Added a regression test in SparkListenerSuite, "removing the last listener is uninterruptible", which removes the last listener (triggering a queue stop) from an already-interrupted thread and asserts that the removal does not throw, the caller's interrupt status is preserved, and the queue is stopped.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    Cursor (Opus 4.8)
    
    Closes #57410 from zeruibao/zeruibao/SPARK-58245.
    
    Authored-by: Zerui Bao <ze...@databricks.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../apache/spark/scheduler/AsyncEventQueue.scala   | 16 ++++++++---
 .../spark/scheduler/SparkListenerSuite.scala       | 31 ++++++++++++++++++++++
 2 files changed, 44 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala b/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
index 3931b8ff5190..735096312b9c 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/AsyncEventQueue.scala
@@ -17,10 +17,11 @@
 
 package org.apache.spark.scheduler
 
-import java.util.concurrent.LinkedBlockingQueue
+import java.util.concurrent.{LinkedBlockingQueue, TimeUnit}
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicLong}
 
 import com.codahale.metrics.{Gauge, Timer}
+import com.google.common.util.concurrent.Uninterruptibles
 
 import org.apache.spark.{SparkConf, SparkContext}
 import org.apache.spark.internal.Logging
@@ -141,7 +142,10 @@ private class AsyncEventQueue(
     }
     if (stopped.compareAndSet(false, true)) {
       eventCount.incrementAndGet()
-      eventQueue.put(POISON_PILL)
+      // Shutdown may run on an interrupted thread, such as a streaming execution thread.
+      // Once stopped is set, the poison pill must still be enqueued or the dispatch thread can
+      // remain blocked forever.
+      Uninterruptibles.putUninterruptibly(eventQueue, POISON_PILL)
     }
     // This thread might be trying to stop itself as part of error handling -- we can't join
     // in that case.
@@ -151,7 +155,13 @@ private class AsyncEventQueue(
       // By default, the `waitingTimeMs` is set to 0,
       // which means it will wait until all events are drained.
       val exitTimeoutMs = conf.get(LISTENER_BUS_EXIT_TIMEOUT)
-      dispatchThread.join(exitTimeoutMs)
+      // Finish waiting despite interruption while preserving the configured timeout.
+      // Uninterruptibles restores the caller's interrupt status before returning.
+      if (exitTimeoutMs == 0) {
+        Uninterruptibles.joinUninterruptibly(dispatchThread)
+      } else {
+        Uninterruptibles.joinUninterruptibly(dispatchThread, exitTimeoutMs, TimeUnit.MILLISECONDS)
+      }
     }
   }
 
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index 6dc4d4da7bfc..b3ec946fb471 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.scheduler
 import java.io.{Externalizable, ObjectInput, ObjectOutput}
 import java.util.{Collections, IdentityHashMap}
 import java.util.concurrent.Semaphore
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
 
 import scala.collection.mutable
 import scala.jdk.CollectionConverters._
@@ -548,6 +549,36 @@ class SparkListenerSuite extends SparkFunSuite with LocalSparkContext with Match
     assert(bus.findListenersByClass[BasicJobCounter]().isEmpty)
   }
 
+  test("removing the last listener is uninterruptible") {
+    val bus = new LiveListenerBus(new SparkConf(false))
+    val counter = new BasicJobCounter()
+    val removalFailure = new AtomicReference[Throwable]()
+    val interruptRestored = new AtomicBoolean(false)
+
+    bus.addToSharedQueue(counter)
+    bus.start(mockSparkContext, mockMetricsSystem)
+
+    val remover = new Thread("interrupted-listener-remover") {
+      override def run(): Unit = {
+        Thread.currentThread().interrupt()
+        try {
+          bus.removeListener(counter)
+        } catch {
+          case error: Throwable => removalFailure.set(error)
+        } finally {
+          interruptRestored.set(Thread.currentThread().isInterrupted)
+        }
+      }
+    }
+    remover.start()
+    remover.join()
+
+    assert(removalFailure.get() == null)
+    assert(interruptRestored.get())
+    assert(bus.activeQueues().isEmpty)
+    bus.stop()
+  }
+
   Seq(true, false).foreach { throwInterruptedException =>
     val suffix = if (throwInterruptedException) "throw interrupt" else "set Thread interrupted"
     test(s"interrupt within listener is handled correctly: $suffix") {


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