You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/08/10 08:58:52 UTC
(spark) branch master updated: [SPARK-58513][CORE] UnifiedMemoryManager wrongly reports INVALID_DRIVER_MEMORY on the executor side
This is an automated email from the ASF dual-hosted git repository.
pan3793 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 f2815e06b39a [SPARK-58513][CORE] UnifiedMemoryManager wrongly reports INVALID_DRIVER_MEMORY on the executor side
f2815e06b39a is described below
commit f2815e06b39a50c3dc87d19f768c9bc3aeac0f52
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Mon Aug 10 16:58:38 2026 +0800
[SPARK-58513][CORE] UnifiedMemoryManager wrongly reports INVALID_DRIVER_MEMORY on the executor side
### What changes were proposed in this pull request?
Make `UnifiedMemoryManager` report role-appropriate errors when the JVM heap is too small:
- `SparkEnv` passes the process role (`isDriver`) to `UnifiedMemoryManager`.
- The system memory check throws `INVALID_DRIVER_MEMORY.SYSTEM_MEMORY` on the driver and the new `INVALID_EXECUTOR_MEMORY.SYSTEM_MEMORY` on executors, each pointing to its own memory option.
- The SPARK-12759 fail-fast check of `spark.executor.memory` is unchanged and runs on both roles, now as `INVALID_EXECUTOR_MEMORY.CONFIG_MEMORY`.
- The two error conditions are restructured with sub-conditions: `INVALID_DRIVER_MEMORY.SYSTEM_MEMORY`, `INVALID_EXECUTOR_MEMORY.{CONFIG_MEMORY, SYSTEM_MEMORY}`.
### Why are the changes needed?
When `spark.executor.memory` is set to 500m, the executor may fail with:
```
Exception in thread "main" java.lang.IllegalArgumentException: System memory 466092032 must be at least 471859200. Please increase heap size using the --driver-memory option or spark.driver.memory in Spark configuration.
```
The message is misleading: the check runs in the executor, whose heap is controlled by `spark.executor.memory`, but it reports driver configuration guidance.
### Does this PR introduce _any_ user-facing change?
Yes, error reporting only; no pass/fail behavior change.
- An executor with insufficient heap now fails with `INVALID_EXECUTOR_MEMORY.SYSTEM_MEMORY`, pointing to `--executor-memory` / `spark.executor.memory`, instead of `INVALID_DRIVER_MEMORY`.
- Existing conditions become sub-conditions: `INVALID_DRIVER_MEMORY` -> `INVALID_DRIVER_MEMORY.SYSTEM_MEMORY`, `INVALID_EXECUTOR_MEMORY` -> `INVALID_EXECUTOR_MEMORY.CONFIG_MEMORY`; messages gain an "Insufficient driver/executor memory:" prefix.
### How was this patch tested?
Added and updated UTs in `UnifiedMemoryManagerSuite`; error definitions are validated by `SparkThrowableSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Generated-by: Claude Code (Claude Fable 5)
Closes #57716 from AngersZhuuuu/SPARK-58513.
Lead-authored-by: Angerszhuuuu <an...@gmail.com>
Co-authored-by: Cheng Pan <ch...@apache.org>
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../src/main/resources/error/error-conditions.json | 28 ++++++++++++--
.../src/main/scala/org/apache/spark/SparkEnv.scala | 5 ++-
.../apache/spark/memory/UnifiedMemoryManager.scala | 31 ++++++++++-----
.../spark/memory/UnifiedMemoryManagerSuite.scala | 44 +++++++++++++++++++---
4 files changed, 89 insertions(+), 19 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 2e6e65865c70..046302512298 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -3947,9 +3947,16 @@
},
"INVALID_DRIVER_MEMORY" : {
"message" : [
- "System memory <systemMemory> must be at least <minSystemMemory>.",
- "Please increase heap size using the --driver-memory option or \"<config>\" in Spark configuration."
+ "Insufficient driver memory:"
],
+ "subClass" : {
+ "SYSTEM_MEMORY" : {
+ "message" : [
+ "System memory <systemMemory> must be at least <minSystemMemory>.",
+ "Please increase heap size using the --driver-memory option or \"<config>\" in Spark configuration."
+ ]
+ }
+ },
"sqlState" : "F0000"
},
"INVALID_EMPTY_LOCATION" : {
@@ -3995,9 +4002,22 @@
},
"INVALID_EXECUTOR_MEMORY" : {
"message" : [
- "Executor memory <executorMemory> must be at least <minSystemMemory>.",
- "Please increase executor memory using the --executor-memory option or \"<config>\" in Spark configuration."
+ "Insufficient executor memory:"
],
+ "subClass" : {
+ "CONFIG_MEMORY" : {
+ "message" : [
+ "Executor memory <executorMemory> must be at least <minSystemMemory>.",
+ "Please increase executor memory using the --executor-memory option or \"<config>\" in Spark configuration."
+ ]
+ },
+ "SYSTEM_MEMORY" : {
+ "message" : [
+ "System memory <systemMemory> must be at least <minSystemMemory>.",
+ "Please increase heap size using the --executor-memory option or \"<config>\" in Spark configuration."
+ ]
+ }
+ },
"sqlState" : "F0000"
},
"INVALID_EXPLODE_EMBEDDED_ARRAY_SCHEMA" : {
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index 8c9be740027f..d48640d9469b 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -512,7 +512,10 @@ class SparkEnv (
} else {
conf.clone.set(MEMORY_OFFHEAP_ENABLED, false).set(MEMORY_OFFHEAP_SIZE, 0L)
}
- _memoryManager = UnifiedMemoryManager(memoryManagerConf, numUsableCores)
+ _memoryManager = UnifiedMemoryManager(
+ memoryManagerConf,
+ numUsableCores,
+ isDriver = SparkContext.isDriver(executorId))
}
}
diff --git a/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala b/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
index 6b278c47f32f..954dc0ae06c4 100644
--- a/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
+++ b/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
@@ -447,7 +447,11 @@ object UnifiedMemoryManager extends Logging {
}
def apply(conf: SparkConf, numCores: Int): UnifiedMemoryManager = {
- val maxMemory = getMaxMemory(conf)
+ apply(conf, numCores, isDriver = true)
+ }
+
+ def apply(conf: SparkConf, numCores: Int, isDriver: Boolean): UnifiedMemoryManager = {
+ val maxMemory = getMaxMemory(conf, isDriver)
new UnifiedMemoryManager(
conf,
maxHeapMemory = maxMemory,
@@ -459,25 +463,34 @@ object UnifiedMemoryManager extends Logging {
/**
* Return the total amount of memory shared between execution and storage, in bytes.
*/
- private def getMaxMemory(conf: SparkConf): Long = {
+ private def getMaxMemory(conf: SparkConf, isDriver: Boolean): Long = {
val systemMemory = conf.get(TEST_MEMORY)
val reservedMemory = conf.getLong(TEST_RESERVED_MEMORY.key,
if (conf.contains(IS_TESTING)) 0 else RESERVED_SYSTEM_MEMORY_BYTES)
val minSystemMemory = (reservedMemory * 1.5).ceil.toLong
if (systemMemory < minSystemMemory) {
- throw new SparkIllegalArgumentException(
- errorClass = "INVALID_DRIVER_MEMORY",
- messageParameters = Map(
- "systemMemory" -> systemMemory.toString,
- "minSystemMemory" -> minSystemMemory.toString,
- "config" -> config.DRIVER_MEMORY.key))
+ if (isDriver) {
+ throw new SparkIllegalArgumentException(
+ errorClass = "INVALID_DRIVER_MEMORY.SYSTEM_MEMORY",
+ messageParameters = Map(
+ "systemMemory" -> systemMemory.toString,
+ "minSystemMemory" -> minSystemMemory.toString,
+ "config" -> config.DRIVER_MEMORY.key))
+ } else {
+ throw new SparkIllegalArgumentException(
+ errorClass = "INVALID_EXECUTOR_MEMORY.SYSTEM_MEMORY",
+ messageParameters = Map(
+ "systemMemory" -> systemMemory.toString,
+ "minSystemMemory" -> minSystemMemory.toString,
+ "config" -> config.EXECUTOR_MEMORY.key))
+ }
}
// SPARK-12759 Check executor memory to fail fast if memory is insufficient
if (conf.contains(config.EXECUTOR_MEMORY)) {
val executorMemory = conf.getSizeAsBytes(config.EXECUTOR_MEMORY.key)
if (executorMemory < minSystemMemory) {
throw new SparkIllegalArgumentException(
- errorClass = "INVALID_EXECUTOR_MEMORY",
+ errorClass = "INVALID_EXECUTOR_MEMORY.CONFIG_MEMORY",
messageParameters = Map(
"executorMemory" -> executorMemory.toString,
"minSystemMemory" -> minSystemMemory.toString,
diff --git a/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala b/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
index 9f0e622b1d51..9a21fa25b35b 100644
--- a/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/memory/UnifiedMemoryManagerSuite.scala
@@ -19,7 +19,7 @@ package org.apache.spark.memory
import org.scalatest.PrivateMethodTester
-import org.apache.spark.SparkConf
+import org.apache.spark.{SparkConf, SparkIllegalArgumentException}
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.Tests._
import org.apache.spark.storage.TestBlockId
@@ -234,10 +234,16 @@ class UnifiedMemoryManagerSuite extends MemoryManagerSuite with PrivateMethodTes
// Try using a system memory that's too small
val conf2 = conf.clone().set(TEST_MEMORY, reservedMemory / 2)
- val exception = intercept[IllegalArgumentException] {
+ val exception = intercept[SparkIllegalArgumentException] {
UnifiedMemoryManager(conf2, numCores = 1)
}
- assert(exception.getMessage.contains("increase heap size"))
+ checkError(
+ exception,
+ condition = "INVALID_DRIVER_MEMORY.SYSTEM_MEMORY",
+ parameters = Map(
+ "systemMemory" -> (reservedMemory / 2).toString,
+ "minSystemMemory" -> (reservedMemory * 1.5).ceil.toLong.toString,
+ "config" -> DRIVER_MEMORY.key))
}
test("insufficient executor memory") {
@@ -253,10 +259,38 @@ class UnifiedMemoryManagerSuite extends MemoryManagerSuite with PrivateMethodTes
// Try using an executor memory that's too small
val conf2 = conf.clone().set(EXECUTOR_MEMORY.key, (reservedMemory / 2).toString)
- val exception = intercept[IllegalArgumentException] {
+ val exception = intercept[SparkIllegalArgumentException] {
UnifiedMemoryManager(conf2, numCores = 1)
}
- assert(exception.getMessage.contains("increase executor memory"))
+ checkError(
+ exception,
+ condition = "INVALID_EXECUTOR_MEMORY.CONFIG_MEMORY",
+ parameters = Map(
+ "executorMemory" -> (reservedMemory / 2).toString,
+ "minSystemMemory" -> (reservedMemory * 1.5).ceil.toLong.toString,
+ "config" -> EXECUTOR_MEMORY.key))
+ }
+
+ test("SPARK-58513: executor validates executor heap") {
+ val systemMemory = 400L * 1024
+ val reservedMemory = 300L * 1024
+ val memoryFraction = 0.8
+ val conf = new SparkConf()
+ .set(MEMORY_FRACTION, memoryFraction)
+ .set(TEST_MEMORY, systemMemory)
+ .set(TEST_RESERVED_MEMORY, reservedMemory)
+ .set(EXECUTOR_MEMORY.key, (500L * 1024).toString)
+
+ val exception = intercept[SparkIllegalArgumentException] {
+ UnifiedMemoryManager(conf, numCores = 1, isDriver = false)
+ }
+ checkError(
+ exception,
+ condition = "INVALID_EXECUTOR_MEMORY.SYSTEM_MEMORY",
+ parameters = Map(
+ "systemMemory" -> systemMemory.toString,
+ "minSystemMemory" -> (reservedMemory * 1.5).ceil.toLong.toString,
+ "config" -> EXECUTOR_MEMORY.key))
}
test("execution can evict cached blocks when there are multiple active tasks (SPARK-12155)") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org