You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2025/11/25 01:08:35 UTC
(spark) branch master updated: [SPARK-54219][CORE] Support `spark.cleaner.referenceTracking.blocking.timeout` config
This is an automated email from the ASF dual-hosted git repository.
dongjoon 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 2316411a19bc [SPARK-54219][CORE] Support `spark.cleaner.referenceTracking.blocking.timeout` config
2316411a19bc is described below
commit 2316411a19bcccf4a0ee9d6fec2f937c44db5584
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Mon Nov 24 17:08:25 2025 -0800
[SPARK-54219][CORE] Support `spark.cleaner.referenceTracking.blocking.timeout` config
### What changes were proposed in this pull request?
SparkContext stop stuck on ContextCleaner
```
25/11/05 18:12:29 ERROR [shutdown-hook-0] ThreadUtils:
14 Driver BLOCKED Blocked by Thread 60 Lock(org.apache.spark.ContextCleaner1726738661})
org.apache.spark.ContextCleaner.stop(ContextCleaner.scala:145)
org.apache.spark.SparkContext.$anonfun$stop$9(SparkContext.scala:2094)
org.apache.spark.SparkContext.$anonfun$stop$9$adapted(SparkContext.scala:2094)
org.apache.spark.SparkContext$$Lambda$5309/807013918.apply(Unknown Source)
scala.Option.foreach(Option.scala:407)
org.apache.spark.SparkContext.$anonfun$stop$8(SparkContext.scala:2094)
org.apache.spark.SparkContext$$Lambda$5308/1445921225.apply$mcV$sp(Unknown Source)
org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1512)
org.apache.spark.SparkContext.stop(SparkContext.scala:2094)
org.apache.spark.SparkContext.stop(SparkContext.scala:2050)
org.apache.spark.sql.SparkSession.stop(SparkSession.scala:718)
com.shopee.data.content.ods.live_performance.Main$.main(Main.scala:62)
com.shopee.data.content.ods.live_performance.Main.main(Main.scala)
sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
java.lang.reflect.Method.invoke(Method.java:498)
org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:751)
```
ContextCleaner stop() will wait lock
```
def stop(): Unit = {
stopped = true
// Interrupt the cleaning thread, but wait until the current task has finished before
// doing so. This guards against the race condition where a cleaning thread may
// potentially clean similarly named variables created by a different SparkContext,
// resulting in otherwise inexplicable block-not-found exceptions (SPARK-6132).
synchronized {
cleaningThread.interrupt()
}
cleaningThread.join()
periodicGCService.shutdown()
}
```
, but one call on keepCleaning() hold the lock
```
25/11/05 18:12:29 ERROR [shutdown-hook-0] ThreadUtils:
60 Spark Context Cleaner TIMED_WAITING Monitor(org.apache.spark.ContextCleaner1726738661})
sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:215)
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedNanos(AbstractQueuedSynchronizer.java:1037)
java.util.concurrent.locks.AbstractQueuedSynchronizer.tryAcquireSharedNanos(AbstractQueuedSynchronizer.java:1328)
scala.concurrent.impl.Promise$DefaultPromise.tryAwait(Promise.scala:248)
scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:258)
scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:263)
org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:294)
org.apache.spark.rpc.RpcTimeout.awaitResult(RpcTimeout.scala:75)
org.apache.spark.storage.BlockManagerMaster.removeBroadcast(BlockManagerMaster.scala:194)
org.apache.spark.broadcast.TorrentBroadcast$.unpersist(TorrentBroadcast.scala:351)
org.apache.spark.broadcast.TorrentBroadcastFactory.unbroadcast(TorrentBroadcastFactory.scala:45)
org.apache.spark.broadcast.BroadcastManager.unbroadcast(BroadcastManager.scala:78)
org.apache.spark.ContextCleaner.doCleanupBroadcast(ContextCleaner.scala:254)
org.apache.spark.ContextCleaner.$anonfun$keepCleaning$3(ContextCleaner.scala:204)
org.apache.spark.ContextCleaner.$anonfun$keepCleaning$3$adapted(ContextCleaner.scala:195)
org.apache.spark.ContextCleaner$$Lambda$1178/1994584033.apply(Unknown Source)
scala.Option.foreach(Option.scala:407)
org.apache.spark.ContextCleaner.$anonfun$keepCleaning$1(ContextCleaner.scala:195) => holding Monitor(org.apache.spark.ContextCleaner1726738661})
org.apache.spark.ContextCleaner$$Lambda$1109/1496842179.apply$mcV$sp(Unknown Source)
org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1474)
org.apache.spark.ContextCleaner.org$apache$spark$ContextCleaner$$keepCleaning(ContextCleaner.scala:189)
org.apache.spark.ContextCleaner$$anon$1.run(ContextCleaner.scala:79)
```
BlockManager stuck on removeBroadcast RpcUtils.INFINITE_TIMEOUT.awaitResult(future) 【PR https://github.com/apache/spark/pull/28924 change here】
```
def removeBroadcast(broadcastId: Long, removeFromMaster: Boolean, blocking: Boolean): Unit = {
val future = driverEndpoint.askSync[Future[Seq[Int]]](
RemoveBroadcast(broadcastId, removeFromMaster))
future.failed.foreach(e =>
logWarning(s"Failed to remove broadcast $broadcastId" +
s" with removeFromMaster = $removeFromMaster - ${e.getMessage}", e)
)(ThreadUtils.sameThread)
if (blocking) {
// the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
RpcUtils.INFINITE_TIMEOUT.awaitResult(future)
}
}
```
For such case only reason should be RPC was missing handling
Driver OOM or A thread leak in yarn nm prevents the creation of new threads to handle RPC.
```
25/11/05 08:16:22 ERROR [metrics-paimon-push-gateway-reporter-2-thread-1] ScheduledReporter: Exception thrown from PushGatewayReporter#report. Exception was suppressed.
java.lang.OutOfMemoryError: unable to create new native thread
at java.lang.Thread.start0(Native Method)
at java.lang.Thread.start(Thread.java:717)
at sun.security.ssl.SSLSocketImpl.readRecord(SSLSocketImpl.java:1115)
at sun.security.ssl.SSLSocketImpl.performInitialHandshake(SSLSocketImpl.java:1388)
at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1416)
at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:1400)
at sun.net.www.protocol.https.HttpsClient.afterConnect(HttpsClient.java:559)
at sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.connect(AbstractDelegateHttpsURLConnection.java:185)
at sun.net.www.protocol.https.HttpsURLConnectionImpl.connect(HttpsURLConnectionImpl.java:167)
at io.prometheus.client.exporter.PushGateway.doRequest(PushGateway.java:243)
at io.prometheus.client.exporter.PushGateway.push(PushGateway.java:134)
at org.apache.paimon.metrics.reporter.PushGatewayReporter.report(PushGatewayReporter.java:84)
at com.codahale.metrics.ScheduledReporter.report(ScheduledReporter.java:253)
at com.codahale.metrics.ScheduledReporter.lambda$start$0(ScheduledReporter.java:182)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
```
For such case we can add a customized wait timeout here to avoid forever stuck
then whole process stuck on here
### Why are the changes needed?
Avoid app stuck
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #52919 from AngersZhuuuu/SPARK-54219.
Authored-by: Angerszhuuuu <an...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../org/apache/spark/internal/config/package.scala | 5 ++
.../apache/spark/storage/BlockManagerMaster.scala | 15 +++---
.../scala/org/apache/spark/util/RpcUtils.scala | 10 ----
.../spark/storage/BlockManagerMasterSuite.scala | 62 ++++++++++++++++++++++
4 files changed, 75 insertions(+), 17 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index e9a3780f0aaa..1e5b2e8018f0 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -1969,6 +1969,11 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT =
+ ConfigBuilder("spark.cleaner.referenceTracking.blocking.timeout")
+ .version("4.2.0")
+ .fallbackConf(Network.NETWORK_TIMEOUT)
+
private[spark] val EXECUTOR_LOGS_ROLLING_STRATEGY =
ConfigBuilder("spark.executor.logs.rolling.strategy")
.version("1.1.0")
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
index e77b4c278574..98bd52fc0886 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
@@ -25,7 +25,8 @@ import org.apache.spark.SparkConf
import org.apache.spark.errors.SparkCoreErrors
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys._
-import org.apache.spark.rpc.RpcEndpointRef
+import org.apache.spark.internal.config.CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT
+import org.apache.spark.rpc.{RpcEndpointRef, RpcTimeout}
import org.apache.spark.storage.BlockManagerMessages._
import org.apache.spark.util.{RpcUtils, ThreadUtils}
@@ -39,6 +40,9 @@ class BlockManagerMaster(
val timeout = RpcUtils.askRpcTimeout(conf)
+ private val waitBlockRemovalTimeout =
+ RpcTimeout(conf, CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT.key, "120s")
+
/** Remove a dead executor from the driver endpoint. This is only called on the driver side. */
def removeExecutor(execId: String): Unit = {
tell(RemoveExecutor(execId))
@@ -195,8 +199,7 @@ class BlockManagerMaster(
log"${MDC(ERROR, e.getMessage)}", e)
)(ThreadUtils.sameThread)
if (blocking) {
- // the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
- RpcUtils.INFINITE_TIMEOUT.awaitResult(future)
+ waitBlockRemovalTimeout.awaitResult(future)
}
}
@@ -208,8 +211,7 @@ class BlockManagerMaster(
log"${MDC(ERROR, e.getMessage)}", e)
)(ThreadUtils.sameThread)
if (blocking) {
- // the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
- RpcUtils.INFINITE_TIMEOUT.awaitResult(future)
+ waitBlockRemovalTimeout.awaitResult(future)
}
}
@@ -223,8 +225,7 @@ class BlockManagerMaster(
log"${MDC(ERROR, e.getMessage)}", e)
)(ThreadUtils.sameThread)
if (blocking) {
- // the underlying Futures will timeout anyway, so it's safe to use infinite timeout here
- RpcUtils.INFINITE_TIMEOUT.awaitResult(future)
+ waitBlockRemovalTimeout.awaitResult(future)
}
}
diff --git a/core/src/main/scala/org/apache/spark/util/RpcUtils.scala b/core/src/main/scala/org/apache/spark/util/RpcUtils.scala
index 58fa4df3b72d..62ae34d5ae27 100644
--- a/core/src/main/scala/org/apache/spark/util/RpcUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/RpcUtils.scala
@@ -17,8 +17,6 @@
package org.apache.spark.util
-import scala.concurrent.duration._
-
import org.apache.spark.SparkConf
import org.apache.spark.internal.config
import org.apache.spark.internal.config.Network._
@@ -55,14 +53,6 @@ private[spark] object RpcUtils {
RpcTimeout(conf, Seq(RPC_LOOKUP_TIMEOUT.key, NETWORK_TIMEOUT.key), "120s")
}
- /**
- * Infinite timeout is used internally, so there's no timeout configuration property that
- * controls it. Therefore, we use "infinite" without any specific reason as its timeout
- * configuration property. And its timeout property should never be accessed since infinite
- * means we never timeout.
- */
- val INFINITE_TIMEOUT = new RpcTimeout(Long.MaxValue.nanos, "infinite")
-
private val MAX_MESSAGE_SIZE_IN_MB = Int.MaxValue / 1024 / 1024
/** Returns the configured max message size for messages in bytes. */
diff --git a/core/src/test/scala/org/apache/spark/storage/BlockManagerMasterSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockManagerMasterSuite.scala
index 2457aef92761..e8aeab13724d 100644
--- a/core/src/test/scala/org/apache/spark/storage/BlockManagerMasterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/storage/BlockManagerMasterSuite.scala
@@ -17,7 +17,17 @@
package org.apache.spark.storage
+import scala.concurrent.{Future, Promise}
+import scala.reflect.ClassTag
+
+import org.mockito.ArgumentMatchers.any
+import org.mockito.Mockito.when
+import org.scalatestplus.mockito.MockitoSugar._
+
import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.internal.config.CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT
+import org.apache.spark.rpc.{RpcEndpointRef, RpcTimeoutException}
+import org.apache.spark.storage.BlockManagerMessages.RemoveRdd
class BlockManagerMasterSuite extends SparkFunSuite {
@@ -30,4 +40,56 @@ class BlockManagerMasterSuite extends SparkFunSuite {
val bmm = new BlockManagerMaster(null, null, new SparkConf, true)
assert(bmm.getStorageStatus.isEmpty)
}
+
+ test("SPARK-54219: wait block removal timeout - success case") {
+ val conf = new SparkConf()
+ conf.set(CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT, 5L)
+
+ val mockDriverEndpoint = mock[RpcEndpointRef]
+ val mockHeartbeatEndpoint = mock[RpcEndpointRef]
+
+ // Create a Future that completes successfully
+ val promise = Promise[Seq[Int]]()
+ val future = promise.future
+
+ // Mock the askSync to return the Future
+ when(
+ mockDriverEndpoint.askSync[Future[Seq[Int]]](any[RemoveRdd])(any[ClassTag[Future[Seq[Int]]]])
+ ).thenReturn(future)
+
+ val bmm = new BlockManagerMaster(mockDriverEndpoint, mockHeartbeatEndpoint, conf, true)
+
+ // Complete the future successfully
+ promise.success(Seq(1, 2, 3))
+
+ // This should not throw an exception
+ bmm.removeRdd(1, blocking = true)
+ }
+
+ test("SPARK-54219: wait block removal timeout - timeout case") {
+ val conf = new SparkConf()
+ // Set a very short timeout (1 second)
+ conf.set(CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT, 1L)
+
+ val mockDriverEndpoint = mock[RpcEndpointRef]
+ val mockHeartbeatEndpoint = mock[RpcEndpointRef]
+
+ // Create a Future that never completes (simulating timeout)
+ val promise = Promise[Seq[Int]]()
+ val future = promise.future
+
+ // Mock the askSync to return the Future
+ when(
+ mockDriverEndpoint.askSync[Future[Seq[Int]]](any[RemoveRdd])(any[ClassTag[Future[Seq[Int]]]])
+ ).thenReturn(future)
+
+ val bmm = new BlockManagerMaster(mockDriverEndpoint, mockHeartbeatEndpoint, conf, true)
+
+ // This should throw RpcTimeoutException
+ val exception = intercept[RpcTimeoutException] {
+ bmm.removeRdd(1, blocking = true)
+ }
+
+ assert(exception.getMessage.contains(CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT.key))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org