You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/11/14 08:29:26 UTC

[GitHub] [spark] pan3793 opened a new pull request, #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

pan3793 opened a new pull request, #38651:
URL: https://github.com/apache/spark/pull/38651

   <!--
   Thanks for sending a pull request!  Here are some tips for you:
     1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
     2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
     3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
     4. Be sure to keep the PR description updated to reflect all changes.
     5. Please write your PR title to summarize what this PR proposes.
     6. If possible, provide a concise example to reproduce the issue for a faster review.
     7. If you want to add a new configuration, please read the guideline first for naming configurations in
        'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
     8. If you want to add or modify an error type or message, please read the guideline first in
        'core/src/main/resources/error/README.md'.
   -->
   
   ### What changes were proposed in this pull request?
   <!--
   Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. 
   If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
     1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
     2. If you fix some SQL features, you can provide some references of other DBMSes.
     3. If there is design documentation, please add the link.
     4. If there is a discussion in the mailing list, please add the link.
   -->
   Shorten graceful shutdown time of `ExecutorPodsSnapshotsStoreImpl#stop` from 30s to 20s to prevent blocking shutdown process
   
   ### Why are the changes needed?
   <!--
   Please clarify why the changes are needed. For instance,
     1. If you propose a new API, clarify the use case for a new API.
     2. If you fix a bug, you can clarify why it is a bug.
   -->
   Currently `ExecutorPodsSnapshotsStoreImpl#stop` has a 30s graceful shutdown duration, since it could be called in shutdown hook, and in default that MGR has 30s shutdown timeout as well, then it may block the following shutdown process.
   
   ```
   SparkContext#stop
     DagScheduler#stop
       TaskScheduler#stop
         KubernetesClusterSchedulerBackend#stop
           ExecutorPodsSnapshotsStoreImpl#stop
   ```
   
   Here is a case, that `SparkContext#stop` is blocked by `ExecutorPodsSnapshotsStoreImpl#stop`, and the interrupt cause the `SparkContext#stop` failed to execute the following shutdown process after `_dagScheduler.stop(exitCode)` including `_eventLogger.foreach(_.stop())`, so that the user can not find the completed application in SHS.
   ```
   2022-11-03 12:51:50 [WARN] [Thread-3] org.apache.hadoop.util.ShutdownHookManager#128 - ShutdownHook '$anon$2' timeout, java.util.concurrent.TimeoutException
   java.util.concurrent.TimeoutException: null
   	at java.util.concurrent.FutureTask.get(FutureTask.java:205) ~[?:1.8.0_345]
   	at org.apache.hadoop.util.ShutdownHookManager.executeShutdown(ShutdownHookManager.java:124) ~[hadoop-common-2.9.2.1.jar:?]
   	at org.apache.hadoop.util.ShutdownHookManager$1.run(ShutdownHookManager.java:95) ~[hadoop-common-2.9.2.1.jar:?]
   2022-11-03 12:51:50 [WARN] [shutdown-hook-0] org.apache.spark.SparkContext#94 - Ignoring Exception while stopping SparkContext from shutdown hook
   java.lang.InterruptedException: null
   	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:2014) ~[?:1.8.0_345]
   	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.awaitNanos(AbstractQueuedSynchronizer.java:2088) ~[?:1.8.0_345]
   	at java.util.concurrent.ThreadPoolExecutor.awaitTermination(ThreadPoolExecutor.java:1475) ~[?:1.8.0_345]
   	at org.apache.spark.util.ThreadUtils$.shutdown(ThreadUtils.scala:347) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.scheduler.cluster.k8s.ExecutorPodsSnapshotsStoreImpl.stop(ExecutorPodsSnapshotsStoreImpl.scala:97) ~[spark-kubernetes_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.scheduler.cluster.k8s.KubernetesClusterSchedulerBackend.$anonfun$stop$2(KubernetesClusterSchedulerBackend.scala:146) ~[spark-kubernetes_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1488) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.scheduler.cluster.k8s.KubernetesClusterSchedulerBackend.stop(KubernetesClusterSchedulerBackend.scala:146) ~[spark-kubernetes_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.scheduler.TaskSchedulerImpl.stop(TaskSchedulerImpl.scala:931) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.scheduler.DAGScheduler.stop(DAGScheduler.scala:2786) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.SparkContext.$anonfun$stop$11(SparkContext.scala:2095) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.util.Utils$.tryLogNonFatalError(Utils.scala:1488) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.SparkContext.stop(SparkContext.scala:2095) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.SparkContext.$anonfun$new$35(SparkContext.scala:660) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.util.SparkShutdownHook.run(ShutdownHookManager.scala:214) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.util.SparkShutdownHookManager.$anonfun$runAll$2(ShutdownHookManager.scala:188) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) ~[scala-library-2.12.15.jar:?]
   	at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:2070) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.util.SparkShutdownHookManager.$anonfun$runAll$1(ShutdownHookManager.scala:188) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23) ~[scala-library-2.12.15.jar:?]
   	at scala.util.Try$.apply(Try.scala:213) ~[scala-library-2.12.15.jar:?]
   	at org.apache.spark.util.SparkShutdownHookManager.runAll(ShutdownHookManager.scala:188) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at org.apache.spark.util.SparkShutdownHookManager$$anon$2.run(ShutdownHookManager.scala:178) ~[spark-core_2.12-3.3.1.1.jar:3.3.1.1]
   	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_345]
   	at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_345]
   	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_345]
   	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_345]
   	at java.lang.Thread.run(Thread.java:750) ~[?:1.8.0_345]
   ```
   
   ### Does this PR introduce _any_ user-facing change?
   <!--
   Note that it means *any* user-facing change including all aspects such as the documentation fix.
   If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
   If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
   If no, write 'No'.
   -->
   No.
   
   ### How was this patch tested?
   <!--
   If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
   If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
   If tests were not added, please describe why they were not added and/or why it was difficult to add.
   If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
   -->
   Existing UT.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023278795


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala:
##########
@@ -723,6 +723,18 @@ private[spark] object Config extends Logging {
       .checkValue(value => value > 0, "Maximum number of pending pods should be a positive integer")
       .createWithDefault(Int.MaxValue)
 
+  val KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD =
+    ConfigBuilder("spark.kubernetes.executorSnapshotsSubscribersShutdownGracePeriod")
+      .doc("Time to wait for graceful shutdown kubernetes-executor-snapshots-subscribers " +
+        "thread pool. Since it may be called by ShutdownHookManager, where timeout is " +
+        "controlled by hadoop configuration `hadoop.service.shutdown.timeout` " +
+        "(default is 30s). As the whole Spark shutdown procedure shares the above timeout, " +
+        "this value should be short than that to prevent blocking the following shutdown " +
+        "procedures.")
+      .version("3.4.0")
+      .timeConf(TimeUnit.SECONDS)

Review Comment:
   Please add `checkValue`.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] tedyu commented on pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
tedyu commented on PR #38651:
URL: https://github.com/apache/spark/pull/38651#issuecomment-1336436604

   Please see https://github.com/apache/spark/pull/38902


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023756963


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala:
##########
@@ -103,7 +103,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val subscribersExecutor = ThreadUtils
       .newDaemonThreadPoolScheduledExecutor(
         "kubernetes-executor-snapshots-subscribers", 2)
-    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(subscribersExecutor)
+    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(sc.conf, subscribersExecutor)

Review Comment:
   yes, but I prefer the SparkConf here, otherwise another backward-compatibility constructor issue will happen if we pass other parameters in the future.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023277805


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,6 +60,7 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,

Review Comment:
   Although this is a `private`, please provide a backward-compatibility in this case by keeping the existing constructor and adding a new one with new parameter `conf`.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] LuciferYang commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1021225985


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -94,7 +95,9 @@ private[spark] class ExecutorPodsSnapshotsStoreImpl(
 
   override def stop(): Unit = {
     pollingTasks.asScala.foreach(_.cancel(false))
-    ThreadUtils.shutdown(subscribersExecutor)
+    // SPARK-41136: it may be called by MGR, and since that MGR has a default 30s timeout, here
+    // a shorter duration 20s is chosen to prevent blocking the following shutdown process.
+    ThreadUtils.shutdown(subscribersExecutor, FiniteDuration(20, TimeUnit.SECONDS))

Review Comment:
   Can this be configurable?
   
   



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] LuciferYang commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023747463


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,10 +60,22 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,
     subscribersExecutor: ScheduledExecutorService,
     clock: Clock = new SystemClock)
   extends ExecutorPodsSnapshotsStore with Logging {
 
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService) = {
+    this(new SparkConf, subscribersExecutor, new SystemClock)
+  }
+
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService,
+      clock: Clock) = {
+    this(new SparkConf, subscribersExecutor, clock)
+  }

Review Comment:
   > I can not merge these two constructers into one,
   > 
   > ```
   >   private[spark] def this(
   >       subscribersExecutor: ScheduledExecutorService,
   >       clock: Clock = new SystemClock) = {
   >     this(new SparkConf, subscribersExecutor, clock)
   >   }
   > ```
   > 
   > it fails compilation
   > 
   > ```
   > [error] /Users/chengpan/Projects/apache-spark/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala:106:64: type mismatch;
   > [error]  found   : org.apache.spark.SparkConf
   > [error]  required: java.util.concurrent.ScheduledExecutorService
   > [error]     val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(sc.conf, subscribersExecutor)
   > [error]
   > ``
   > ```
   
   Need remove the default value of `SystemClock` from primary constructor if you want to merge them to one



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023621416


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala:
##########
@@ -103,7 +103,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val subscribersExecutor = ThreadUtils
       .newDaemonThreadPoolScheduledExecutor(
         "kubernetes-executor-snapshots-subscribers", 2)
-    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(subscribersExecutor)
+    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(sc.conf, subscribersExecutor)

Review Comment:
   BTW, it seems that we don't need to hand over the whole `SparkConf` here. What we need is only `sc.conf.get(KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD)`, isnt' it?



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] LuciferYang commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023764590


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,6 +60,7 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,

Review Comment:
   If `conf` is placed after `clock` and with a default value `newSparkConf` will there be compatibility problems? If not, we can delete the other two constructors
   
   



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1022306648


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -94,7 +95,9 @@ private[spark] class ExecutorPodsSnapshotsStoreImpl(
 
   override def stop(): Unit = {
     pollingTasks.asScala.foreach(_.cancel(false))
-    ThreadUtils.shutdown(subscribersExecutor)
+    // SPARK-41136: it may be called by MGR, and since that MGR has a default 30s timeout, here

Review Comment:
   - In this case, we don't need JIRA id in the code. 
   - Please write full name instead of MGR.
   - When MGR has non-default timeout like 20s, what happens?



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023762475


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,10 +60,22 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,
     subscribersExecutor: ScheduledExecutorService,
     clock: Clock = new SystemClock)
   extends ExecutorPodsSnapshotsStore with Logging {
 
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService) = {
+    this(new SparkConf, subscribersExecutor, new SystemClock)
+  }
+
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService,
+      clock: Clock) = {
+    this(new SparkConf, subscribersExecutor, clock)
+  }

Review Comment:
   Then `clock` is required if user wants to pass `conf`, it's not friendly IMO.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on PR #38651:
URL: https://github.com/apache/spark/pull/38651#issuecomment-1318542102

   @dongjoon-hyun @LuciferYang would you please take a look again?


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1024775551


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,6 +60,7 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,

Review Comment:
   I moved `conf: SparkConf = new SparkConf` to the last as suggested



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023614721


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,10 +60,22 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,
     subscribersExecutor: ScheduledExecutorService,
     clock: Clock = new SystemClock)
   extends ExecutorPodsSnapshotsStore with Logging {
 
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService) = {
+    this(new SparkConf, subscribersExecutor, new SystemClock)
+  }
+
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService,
+      clock: Clock) = {
+    this(new SparkConf, subscribersExecutor, clock)
+  }

Review Comment:
   Oh, interesting.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] AmplabJenkins commented on pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #38651:
URL: https://github.com/apache/spark/pull/38651#issuecomment-1313645025

   Can one of the admins verify this patch?


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1022312510


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -94,7 +95,9 @@ private[spark] class ExecutorPodsSnapshotsStoreImpl(
 
   override def stop(): Unit = {
     pollingTasks.asScala.foreach(_.cancel(false))
-    ThreadUtils.shutdown(subscribersExecutor)
+    // SPARK-41136: it may be called by MGR, and since that MGR has a default 30s timeout, here

Review Comment:
   Thanks, will make the hardcoded 20s configurable, and update and move the comment into 
   configuration description.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] holdenk commented on pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
holdenk commented on PR #38651:
URL: https://github.com/apache/spark/pull/38651#issuecomment-1336407949

   We don't normally do inline diff views, could you raise a PR?


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023477562


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,10 +60,22 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,
     subscribersExecutor: ScheduledExecutorService,
     clock: Clock = new SystemClock)
   extends ExecutorPodsSnapshotsStore with Logging {
 
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService) = {
+    this(new SparkConf, subscribersExecutor, new SystemClock)
+  }
+
+  private[spark] def this(
+      subscribersExecutor: ScheduledExecutorService,
+      clock: Clock) = {
+    this(new SparkConf, subscribersExecutor, clock)
+  }

Review Comment:
   I can not merge these two constructers into one,
   ```
     private[spark] def this(
         subscribersExecutor: ScheduledExecutorService,
         clock: Clock = new SystemClock) = {
       this(new SparkConf, subscribersExecutor, clock)
     }
   ```
   it fails compilation
   ```
   [error] /Users/chengpan/Projects/apache-spark/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala:106:64: type mismatch;
   [error]  found   : org.apache.spark.SparkConf
   [error]  required: java.util.concurrent.ScheduledExecutorService
   [error]     val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(sc.conf, subscribersExecutor)
   [error]
   ``



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023621416


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala:
##########
@@ -103,7 +103,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val subscribersExecutor = ThreadUtils
       .newDaemonThreadPoolScheduledExecutor(
         "kubernetes-executor-snapshots-subscribers", 2)
-    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(subscribersExecutor)
+    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(sc.conf, subscribersExecutor)

Review Comment:
   BTW, it seems that we don't need to hand over the whole `SparkConf` here. What we need is only `conf.get(KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD)`, isnt' it?



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] LuciferYang commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023764590


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,6 +60,7 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,

Review Comment:
   If `conf` is placed after `clock` and with a default value `new SparkConf` will there be compatibility problems? If not, we can delete the other two constructors
   
   



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] dongjoon-hyun closed pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun closed pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process
URL: https://github.com/apache/spark/pull/38651


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023478697


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -57,6 +60,7 @@ import org.apache.spark.util.ThreadUtils
  * The subscriber notification callback is guaranteed to be called from a single thread at a time.
  */
 private[spark] class ExecutorPodsSnapshotsStoreImpl(
+    conf: SparkConf,

Review Comment:
   the backward-compatible constructors were added



##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala:
##########
@@ -723,6 +723,18 @@ private[spark] object Config extends Logging {
       .checkValue(value => value > 0, "Maximum number of pending pods should be a positive integer")
       .createWithDefault(Int.MaxValue)
 
+  val KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD =
+    ConfigBuilder("spark.kubernetes.executorSnapshotsSubscribersShutdownGracePeriod")
+      .doc("Time to wait for graceful shutdown kubernetes-executor-snapshots-subscribers " +
+        "thread pool. Since it may be called by ShutdownHookManager, where timeout is " +
+        "controlled by hadoop configuration `hadoop.service.shutdown.timeout` " +
+        "(default is 30s). As the whole Spark shutdown procedure shares the above timeout, " +
+        "this value should be short than that to prevent blocking the following shutdown " +
+        "procedures.")
+      .version("3.4.0")
+      .timeConf(TimeUnit.SECONDS)

Review Comment:
   added



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on PR #38651:
URL: https://github.com/apache/spark/pull/38651#issuecomment-1316286965

   @dongjoon-hyun thanks for review, I addressed your comments.


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1023756963


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterManager.scala:
##########
@@ -103,7 +103,7 @@ private[spark] class KubernetesClusterManager extends ExternalClusterManager wit
     val subscribersExecutor = ThreadUtils
       .newDaemonThreadPoolScheduledExecutor(
         "kubernetes-executor-snapshots-subscribers", 2)
-    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(subscribersExecutor)
+    val snapshotsStore = new ExecutorPodsSnapshotsStoreImpl(sc.conf, subscribersExecutor)

Review Comment:
   yes, but I prefer the SparkConf here, otherwise another in-backward-compatibility constructor issue will happen if we pass other parameters in the future.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] pan3793 commented on a diff in pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
pan3793 commented on code in PR #38651:
URL: https://github.com/apache/spark/pull/38651#discussion_r1021830231


##########
resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala:
##########
@@ -94,7 +95,9 @@ private[spark] class ExecutorPodsSnapshotsStoreImpl(
 
   override def stop(): Unit = {
     pollingTasks.asScala.foreach(_.cancel(false))
-    ThreadUtils.shutdown(subscribersExecutor)
+    // SPARK-41136: it may be called by MGR, and since that MGR has a default 30s timeout, here
+    // a shorter duration 20s is chosen to prevent blocking the following shutdown process.
+    ThreadUtils.shutdown(subscribersExecutor, FiniteDuration(20, TimeUnit.SECONDS))

Review Comment:
   yes, will update it tomorrow.



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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

[GitHub] [spark] tedyu commented on pull request #38651: [SPARK-41136][K8S] Shorten graceful shutdown time of ExecutorPodsSnapshotsStoreImpl to prevent blocking shutdown process

Posted by GitBox <gi...@apache.org>.
tedyu commented on PR #38651:
URL: https://github.com/apache/spark/pull/38651#issuecomment-1336401312

   I think we can do more than setting default value for `KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD`.
   
   What do you think of the following change ?
   
   Thanks
   ```
   diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala
   index 49ab1d3248..bac171336a 100644
   --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala
   +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsSnapshotsStoreImpl.scala
   @@ -29,6 +29,7 @@ import scala.util.control.NonFatal
    import io.fabric8.kubernetes.api.model.Pod
   
    import org.apache.spark.{SparkConf, SparkContext}
   +import org.apache.spark.deploy.SparkHadoopUtil
    import org.apache.spark.deploy.k8s.Config.KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD
    import org.apache.spark.internal.Logging
    import org.apache.spark.util.Clock
   @@ -98,7 +99,15 @@ private[spark] class ExecutorPodsSnapshotsStoreImpl(
   
      override def stop(): Unit = {
        pollingTasks.asScala.foreach(_.cancel(false))
   -    val awaitSeconds = conf.get(KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD)
   +    var awaitSeconds = conf.get(KUBERNETES_EXECUTOR_SNAPSHOTS_SUBSCRIBERS_GRACE_PERIOD)
   +    val hadoopConf = SparkHadoopUtil.get.newConfiguration(conf)
   +    if (hadoopConf.get("hadoop.service.shutdown.timeout") != null) {
   +        val hadoopTimeout = hadoopConf.get("hadoop.service.shutdown.timeout").toLong
   +        val GRACE = 8
   +        if (hadoopTimeout-GRACE <= awaitSeconds) {
   +            awaitSeconds = hadoopTimeout-GRACE
   +        }
   +    }
        ThreadUtils.shutdown(subscribersExecutor, FiniteDuration(awaitSeconds, TimeUnit.SECONDS))
      }
   ```


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


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