This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 2292585c0b3c [SPARK-57867][CORE] Driver should not reserve off-heap
memory in non-local mode
2292585c0b3c is described below
commit 2292585c0b3cc6c4d94271205b95b7ba2f0a1a5f
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Sun Jul 5 10:39:35 2026 -0700
[SPARK-57867][CORE] Driver should not reserve off-heap memory in non-local
mode
### What changes were proposed in this pull request?
This PR proposes to stop reserving off-heap memory pools
(`spark.memory.offHeap.size`) in the driver's `MemoryManager` in non-`local`
deployments. `SparkEnv.initializeMemoryManager` takes a new `offHeapAllowed`
parameter, and `SparkContext` passes `offHeapAllowed = isLocal` for the driver.
The executor path and `local` mode are unchanged.
### Why are the changes needed?
Off-heap memory is accounted for only in **executor** resource sizing
(`ResourceProfile.OFFHEAP_MEM`, YARN executor container size, K8s
`BasicExecutorFeatureStep`). The driver's container memory request never
includes `spark.memory.offHeap.size`. So, we should not allow it.
However, with `spark.memory.offHeap.enabled=true`, the Executors UI and
REST API show the driver with `spark.memory.offHeap.size` of `Off Heap Storage
Memory` like the following, which is very misleading.
**BEFORE**
<img width="612" height="243" alt="Screenshot 2026-07-01 at 15 24 59"
src="https://github.com/user-attachments/assets/2ec8e6d6-9654-4914-8e18-afa100802962";
/>
**AFTER**
<img width="621" height="235" alt="Screenshot 2026-07-01 at 15 23 11"
src="https://github.com/user-attachments/assets/94157da3-2e34-4a23-a46a-f07474131a41";
/>
### Does this PR introduce _any_ user-facing change?
No. The driver in non-local deployments never uses it: it runs no tasks,
stores no off-heap blocks.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #56945 from dongjoon-hyun/SPARK-57867.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../main/scala/org/apache/spark/SparkContext.scala | 6 +++++-
core/src/main/scala/org/apache/spark/SparkEnv.scala | 11 +++++++++--
.../scala/org/apache/spark/SparkContextSuite.scala | 20 ++++++++++++++++++++
.../spark/internal/plugin/PluginContainerSuite.scala | 5 +++--
4 files changed, 37 insertions(+), 5 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala
b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 0a701e1967bd..3610c676c40f 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -590,7 +590,11 @@ class SparkContext(config: SparkConf) extends Logging {
_plugins = PluginContainer(this, _resources.asJava)
_resourceProfileManager = new ResourceProfileManager(_conf, _listenerBus)
_env.initializeShuffleManager()
- _env.initializeMemoryManager(SparkContext.numDriverCores(master, conf))
+ // The driver in non-local deployments never runs tasks or stores off-heap
blocks, and its
+ // container memory is not sized for spark.memory.offHeap.size, so don't
reserve off-heap
+ // memory there.
+ _env.initializeMemoryManager(
+ SparkContext.numDriverCores(master, conf), offHeapAllowed = isLocal)
// Create and start the scheduler
val (sched, ts) = SparkContext.createTaskScheduler(this, master)
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala
b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index 39f12e03a933..2bc10adcbe3a 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -347,10 +347,17 @@ class SparkEnv (
}
}
- private[spark] def initializeMemoryManager(numUsableCores: Int): Unit = {
+ private[spark] def initializeMemoryManager(
+ numUsableCores: Int,
+ offHeapAllowed: Boolean = true): Unit = {
Preconditions.checkState(null == memoryManager,
"Memory manager already initialized to %s", _memoryManager)
- _memoryManager = UnifiedMemoryManager(conf, numUsableCores)
+ val memoryManagerConf = if (offHeapAllowed) {
+ conf
+ } else {
+ conf.clone.set(MEMORY_OFFHEAP_ENABLED, false).set(MEMORY_OFFHEAP_SIZE,
0L)
+ }
+ _memoryManager = UnifiedMemoryManager(memoryManagerConf, numUsableCores)
}
}
diff --git a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
index 36a1a1d07daa..517506b0f26f 100644
--- a/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
+++ b/core/src/test/scala/org/apache/spark/SparkContextSuite.scala
@@ -1481,6 +1481,26 @@ class SparkContextSuite extends SparkFunSuite with
LocalSparkContext with Eventu
}.getMessage
assert(m.contains("Number of cores to allocate for each task should be
positive."))
}
+
+ test("SPARK-57867: Driver should not reserve off-heap memory in non-local
mode") {
+ val conf = new SparkConf()
+ .setAppName("test")
+ .setMaster("local-cluster[1,1,1024]")
+ .set(MEMORY_OFFHEAP_ENABLED, true)
+ .set(MEMORY_OFFHEAP_SIZE, 5L * 1024 * 1024)
+ sc = new SparkContext(conf)
+ assert(sc.env.memoryManager.maxOffHeapStorageMemory === 0)
+ }
+
+ test("SPARK-57867: Driver should reserve off-heap memory in local mode") {
+ val conf = new SparkConf()
+ .setAppName("test")
+ .setMaster("local")
+ .set(MEMORY_OFFHEAP_ENABLED, true)
+ .set(MEMORY_OFFHEAP_SIZE, 5L * 1024 * 1024)
+ sc = new SparkContext(conf)
+ assert(sc.env.memoryManager.maxOffHeapStorageMemory > 0)
+ }
}
object SparkContextSuite {
diff --git
a/core/src/test/scala/org/apache/spark/internal/plugin/PluginContainerSuite.scala
b/core/src/test/scala/org/apache/spark/internal/plugin/PluginContainerSuite.scala
index 700a17649b76..b1c5683cb225 100644
---
a/core/src/test/scala/org/apache/spark/internal/plugin/PluginContainerSuite.scala
+++
b/core/src/test/scala/org/apache/spark/internal/plugin/PluginContainerSuite.scala
@@ -244,8 +244,9 @@ class PluginContainerSuite extends SparkFunSuite with
LocalSparkContext {
sc = new SparkContext(conf)
val memoryManager = sc.env.memoryManager
- assert(memoryManager.tungstenMemoryMode == MemoryMode.OFF_HEAP)
- assert(memoryManager.maxOffHeapStorageMemory ==
MemoryOverridePlugin.offHeapMemory)
+ // SPARK-57867: The driver does not reserve off-heap memory in non-local
mode
+ assert(memoryManager.tungstenMemoryMode == MemoryMode.ON_HEAP)
+ assert(memoryManager.maxOffHeapStorageMemory == 0)
val defaultResourceProfile =
sc.resourceProfileManager.defaultResourceProfile
assert(512L ==
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
