[SPARK-51436][SQL] Change the mayInterruptIfRunning from true to false

apache · Mar 8, 2025 · a6dbeae · a6dbeae
1 parent c0ec84e
commit a6dbeae
Show file tree

Hide file tree

Showing 8 changed files with 11 additions and 11 deletions.
diff --git a/...afka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/FetchedDataPool.scala b/...afka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/consumer/FetchedDataPool.scala
@@ -139,7 +139,7 @@ private[consumer] class FetchedDataPool(
   }
 
   def reset(): Unit = synchronized {
-    scheduled.foreach(_.cancel(true))
+    scheduled.foreach(_.cancel(false))
 
     cache.clear()
     numTotalElements.reset()

diff --git a/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala b/core/src/main/scala/org/apache/spark/HeartbeatReceiver.scala
@@ -250,7 +250,7 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
 
   override def onStop(): Unit = {
     if (timeoutCheckingTask != null) {
-      timeoutCheckingTask.cancel(true)
+      timeoutCheckingTask.cancel(false)
     }
     eventLoopThread.shutdownNow()
     killExecutorThread.shutdownNow()

diff --git a/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala b/core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala
@@ -277,7 +277,7 @@ private[spark] class StandaloneAppClient(
 
     override def onStop(): Unit = {
       if (registrationRetryTimer.get != null) {
-        registrationRetryTimer.get.cancel(true)
+        registrationRetryTimer.get.cancel(false)
       }
       registrationRetryThread.shutdownNow()
       registerMasterFutures.get.foreach(_.cancel(true))

diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala
@@ -214,10 +214,10 @@ private[deploy] class Master(
     applicationMetricsSystem.report()
     // prevent the CompleteRecovery message sending to restarted master
     if (recoveryCompletionTask != null) {
-      recoveryCompletionTask.cancel(true)
+      recoveryCompletionTask.cancel(false)
     }
     if (checkForWorkerTimeOutTask != null) {
-      checkForWorkerTimeOutTask.cancel(true)
+      checkForWorkerTimeOutTask.cancel(false)
     }
     forwardMessageThread.shutdownNow()
     webUi.stop()

diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala b/core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
@@ -403,7 +403,7 @@ private[deploy] class Worker(
         // We have exceeded the initial registration retry threshold
         // All retries from now on should use a higher interval
         if (connectionAttemptCount == INITIAL_REGISTRATION_RETRIES) {
-          registrationRetryTimer.foreach(_.cancel(true))
+          registrationRetryTimer.foreach(_.cancel(false))
           registrationRetryTimer = Some(
             forwardMessageScheduler.scheduleAtFixedRate(
               () => Utils.tryLogNonFatalError { self.send(ReregisterWithMaster) },
@@ -426,7 +426,7 @@ private[deploy] class Worker(
       registerMasterFutures.foreach(_.cancel(true))
       registerMasterFutures = null
     }
-    registrationRetryTimer.foreach(_.cancel(true))
+    registrationRetryTimer.foreach(_.cancel(false))
     registrationRetryTimer = None
   }
 

diff --git a/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala b/core/src/main/scala/org/apache/spark/rpc/netty/NettyRpcEnv.scala
@@ -269,7 +269,7 @@ private[netty] class NettyRpcEnv(
         }
       }, timeout.duration.toNanos, TimeUnit.NANOSECONDS)
       promise.future.onComplete { v =>
-        timeoutCancelable.cancel(true)
+        timeoutCancelable.cancel(false)
       }(ThreadUtils.sameThread)
     } catch {
       case NonFatal(e) =>

diff --git a/...main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala b/...main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsPollingSnapshotSource.scala
@@ -63,7 +63,7 @@ class ExecutorPodsPollingSnapshotSource(
   @Since("3.1.3")
   def stop(): Unit = {
     if (pollingFuture != null) {
-      pollingFuture.cancel(true)
+      pollingFuture.cancel(false)
       pollingFuture = null
     }
     ThreadUtils.shutdown(pollingExecutor)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala
@@ -108,7 +108,7 @@ trait BroadcastExchangeLike extends Exchange {
         case Some(r) => sparkContext.cancelJobsWithTag(this.jobTag, r)
         case None => sparkContext.cancelJobsWithTag(this.jobTag)
       }
-      this.relationFuture.cancel(true)
+      this.relationFuture.cancel(false)
     }
   }
 
@@ -257,7 +257,7 @@ case class BroadcastExchangeExec(
         logError(log"Could not execute broadcast in ${MDC(TIMEOUT, timeout)} secs.", ex)
         if (!relationFuture.isDone) {
           sparkContext.cancelJobsWithTag(jobTag, "The corresponding broadcast query has failed.")
-          relationFuture.cancel(true)
+          relationFuture.cancel(false)
         }
         throw QueryExecutionErrors.executeBroadcastTimeoutError(timeout, Some(ex))
     }