This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 337eb8546a53 [SPARK-53128][CORE] Include unmanaged memory bytes in the
usage log before execution memory OOM
337eb8546a53 is described below
commit 337eb8546a53f80706fcaf609bb4d1d7696f36b9
Author: Hongze Zhang <[email protected]>
AuthorDate: Mon Nov 3 13:49:03 2025 -0800
[SPARK-53128][CORE] Include unmanaged memory bytes in the usage log before
execution memory OOM
### What changes were proposed in this pull request?
We have a log before OOM for off-heap memory allocation.
Before the change, the log is:
> 25/08/05 16:44:32 INFO TaskMemoryManager: 100 bytes of memory are used
for execution and 100 bytes of memory are used for storage
After:
> 25/08/05 16:44:32 INFO TaskMemoryManager: 100 bytes of memory are used
for execution and 100 bytes of memory are used for storage and 500 bytes of
memory are used but unmanaged
### Why are the changes needed?
Following https://github.com/apache/spark/pull/51708, to allow user to know
the reason if the unmanaged memory causes OOM.
### Does this PR introduce _any_ user-facing change?
Only changes a log message.
### How was this patch tested?
Existing tests.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #51848 from zhztheplayer/wip-53128.
Authored-by: Hongze Zhang <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit c4ad38117e4be3fed7534e0159c1203a01f058e1)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java | 7 +++++--
.../main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala | 8 ++++++++
2 files changed, 13 insertions(+), 2 deletions(-)
diff --git a/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
b/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
index ab9e470e0c2c..8b41df6b269f 100644
--- a/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
+++ b/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
@@ -339,9 +339,12 @@ public class TaskMemoryManager {
MDC.of(LogKeys.MEMORY_SIZE, memoryNotAccountedFor),
MDC.of(LogKeys.TASK_ATTEMPT_ID, taskAttemptId));
logger.info(
- "{} bytes of memory are used for execution and {} bytes of memory are
used for storage",
+ "{} bytes of memory are used for execution " +
+ "and {} bytes of memory are used for storage " +
+ "and {} bytes of unmanaged memory are used",
MDC.of(LogKeys.EXECUTION_MEMORY_SIZE,
memoryManager.executionMemoryUsed()),
- MDC.of(LogKeys.STORAGE_MEMORY_SIZE,
memoryManager.storageMemoryUsed()));
+ MDC.of(LogKeys.STORAGE_MEMORY_SIZE,
memoryManager.storageMemoryUsed()),
+ MDC.of(LogKeys.MEMORY_SIZE,
UnifiedMemoryManager$.MODULE$.getUnmanagedMemoryUsed()));
}
}
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 db51f14415e1..6b278c47f32f 100644
--- a/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
+++ b/core/src/main/scala/org/apache/spark/memory/UnifiedMemoryManager.scala
@@ -273,6 +273,14 @@ object UnifiedMemoryManager extends Logging {
// Atomic flag to ensure polling is only started once per JVM
private val pollingStarted = new AtomicBoolean(false)
+ /**
+ * Returns the total unmanaged memory in bytes, including both
+ * on-heap unmanaged memory and off-heap unmanaged memory.
+ */
+ private[spark] def getUnmanagedMemoryUsed: Long = {
+ UnifiedMemoryManager.unmanagedOnHeapUsed.get() +
UnifiedMemoryManager.unmanagedOffHeapUsed.get()
+ }
+
/**
* Register an unmanaged memory consumer to track its memory usage.
*
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
