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 2026/09/08 13:36:23 UTC
(spark) branch branch-4.x updated: [SPARK-59331][K8S] Turn off recovery mode when a held application resumes
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new bb7a15edb28c [SPARK-59331][K8S] Turn off recovery mode when a held application resumes
bb7a15edb28c is described below
commit bb7a15edb28cf9e1f95ee5688ee78250b837e08f
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Tue Sep 8 06:36:03 2026 -0700
[SPARK-59331][K8S] Turn off recovery mode when a held application resumes
### What changes were proposed in this pull request?
This PR proposes to turn recovery mode off when a held application resumes, so that it allocates normal executors again.
- `ExecutorPodsAllocator.unsetRecoveryMode()` removes `spark.kubernetes.allocation.recoveryMode.enabled` from the driver conf if the driver turned it on after an OOM. A value configured by the user is left untouched.
- `KubernetesClusterSchedulerBackend.setExecutorsHeld(false)` calls it, which is the hook `SparkContext.resumeExecutors()` uses.
- `running-on-kubernetes.md` is updated accordingly.
### Why are the changes needed?
Recovery mode stays on for the rest of the driver's lifetime once an OOM is detected. A hold drains every executor, including the recovery-mode ones, and a resume starts allocating from scratch, so it is the natural point to start over with the configured executor cores. Otherwise a long-running driver, e.g. a Spark Connect server, stays on single-task executors after a single OOM until it is restarted.
### Does this PR introduce _any_ user-facing change?
No for the released versions, since hold and resume (4.4.0) is not released yet. Within the unreleased branches, a K8s application that entered recovery mode after an OOM allocates normal executors again after a hold and resume. An explicitly configured `spark.kubernetes.allocation.recoveryMode.enabled=true` stays on as before.
### How was this patch tested?
Pass the CIs with the newly added test cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5.1
Closes #58618 from dongjoon-hyun/SPARK-59331.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit b017da6f98a34490dba262e801e851dbf7844a0c)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
docs/running-on-kubernetes.md | 13 ++++++++-----
.../scheduler/cluster/k8s/ExecutorPodsAllocator.scala | 18 ++++++++++++++++++
.../k8s/KubernetesClusterSchedulerBackend.scala | 11 +++++++++++
.../cluster/k8s/ExecutorPodsAllocatorSuite.scala | 19 +++++++++++++++++++
.../k8s/KubernetesClusterSchedulerBackendSuite.scala | 7 +++++++
5 files changed, 63 insertions(+), 5 deletions(-)
diff --git a/docs/running-on-kubernetes.md b/docs/running-on-kubernetes.md
index ad9feb3b7e85..00f244d6307f 100644
--- a/docs/running-on-kubernetes.md
+++ b/docs/running-on-kubernetes.md
@@ -524,16 +524,19 @@ single task at a time, while its pod still requests the configured executor core
memory-hungry task gets the whole executor memory to itself instead of sharing it with other tasks,
which lets the remaining tasks and stages finish instead of failing repeatedly with the same OOM.
-The property is unset by default, so recovery mode is off until the driver detects the first OOM.
-At that point the driver turns it on and it stays on for the rest of the driver's lifetime. Executors
-that were already running are not changed. Recovery-mode executors always derive their announced cores
-from `spark.task.cpus`, not from the resource profile of the stage. To disable the automatic switch:
+The property is unset by default, so recovery mode is off until the driver detects the first OOM. At
+that point the driver turns it on, and it stays on until the application is held and resumed: the hold
+drains every executor the OOM was reacting to, so the resumed application starts over with normal
+executors. Executors that were already running are not changed. Recovery-mode executors always derive
+their announced cores from `spark.task.cpus`, not from the resource profile of the stage. To disable
+the automatic switch:
```
--conf spark.kubernetes.allocation.recoveryMode.enabled=false
```
-Setting the property to `true` starts the application in recovery mode from the beginning. See the
+Setting the property to `true` starts the application in recovery mode from the beginning, and a hold
+and resume does not turn it off, because only the driver's own reaction to an OOM is reverted. See the
description of `spark.kubernetes.allocation.recoveryMode.enabled` in the
[Spark Properties](#spark-properties) table for the behavior when `spark.task.cpus` is 0.5 or less.
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
index 2523cd64d3d7..3a3b4f0afa82 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala
@@ -469,11 +469,29 @@ class ExecutorPodsAllocator(
}
}
+ // Whether the user configured recovery mode. Only the switch an OOM makes is reverted when
+ // the application resumes: an application configured to run in recovery mode stays in it.
+ private val recoveryModeConfigured =
+ conf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).isDefined
+
def setRecoveryMode(): Unit = {
conf.setIfMissing(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED, true)
warnIfRecoveryModeCannotIsolateSingleTask()
}
+ /**
+ * Leave the recovery mode an OOM turned on, so that the executors created afterwards are
+ * normal executors again. Called when a held application resumes: the hold drained the
+ * executors the OOM was reacting to, and the resumed application starts over.
+ */
+ def unsetRecoveryMode(): Unit = {
+ if (!recoveryModeConfigured &&
+ conf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).isDefined) {
+ conf.remove(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED)
+ logInfo("Recovery mode is turned off, so the executors created next are normal executors.")
+ }
+ }
+
// Recovery mode's single-task guarantee is enforced by announcing SPARK_EXECUTOR_CORES =
// ceil(spark.task.cpus), which must be a whole number. When the task cpus amount is 0.5 or
// less, the single announced core fits more than one task, so the guarantee cannot hold;
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
index 27a0b320cbf9..6c32bed33218 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala
@@ -203,6 +203,17 @@ private[spark] class KubernetesClusterSchedulerBackend(
conf.get(KUBERNETES_ALLOCATION_PODS_ALLOCATOR) == "direct"
}
+ // A hold drains every executor, including the recovery-mode ones that replaced an
+ // OOM-terminated executor, so a resumed application starts over with normal executors
+ // instead of staying in the recovery mode the OOM turned on.
+ private[spark] override def setExecutorsHeld(held: Boolean): Unit = {
+ super.setExecutorsHeld(held)
+ podAllocator match {
+ case allocator: ExecutorPodsAllocator if !held => allocator.unsetRecoveryMode()
+ case _ => // no-op
+ }
+ }
+
override def sufficientResourcesRegistered(): Boolean = {
totalRegisteredExecutors.get() >= minRegisteredExecutors
}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
index a3e0b6625ce4..8344e82f908d 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala
@@ -1133,4 +1133,23 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
podsAllocator.setRecoveryMode()
assert(!newConf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).get)
}
+
+ test("SPARK-59331: unsetRecoveryMode leaves the recovery mode that setRecoveryMode turned on") {
+ val newConf = conf.clone
+ val podsAllocator = new ExecutorPodsAllocator(newConf, secMgr, executorBuilder,
+ kubernetesClient, snapshotsStore, waitForExecutorPodsClock)
+ podsAllocator.setRecoveryMode()
+ assert(newConf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).contains(true))
+ podsAllocator.unsetRecoveryMode()
+ assert(newConf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).isEmpty)
+ }
+
+ test("SPARK-59331: unsetRecoveryMode keeps a configured recovery mode") {
+ val newConf = conf.clone.set(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED, true)
+ val podsAllocator = new ExecutorPodsAllocator(newConf, secMgr, executorBuilder,
+ kubernetesClient, snapshotsStore, waitForExecutorPodsClock)
+ podsAllocator.setRecoveryMode()
+ podsAllocator.unsetRecoveryMode()
+ assert(newConf.get(KUBERNETES_ALLOCATION_RECOVERY_MODE_ENABLED).contains(true))
+ }
}
diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
index 64734cc6c612..9abefbe8e01c 100644
--- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
+++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala
@@ -510,4 +510,11 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn
}
assert(schedulerBackendUnderTest.supportsExecutorHold)
}
+
+ test("SPARK-59331: resuming a held application leaves the recovery mode an OOM turned on") {
+ schedulerBackendUnderTest.setExecutorsHeld(true)
+ verify(podAllocator, never()).unsetRecoveryMode()
+ schedulerBackendUnderTest.setExecutorsHeld(false)
+ verify(podAllocator).unsetRecoveryMode()
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org