You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/11 02:37:57 UTC
(spark) branch master updated: [SPARK-58505][CORE] Attach a bounded per-consumer memory breakdown to UNABLE_TO_ACQUIRE_MEMORY errors
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 32e9ac5324d4 [SPARK-58505][CORE] Attach a bounded per-consumer memory breakdown to UNABLE_TO_ACQUIRE_MEMORY errors
32e9ac5324d4 is described below
commit 32e9ac5324d49245e1ab1bfd0aa609b9758602b0
Author: Ganesha S <ga...@gmail.com>
AuthorDate: Tue Aug 11 10:37:42 2026 +0800
[SPARK-58505][CORE] Attach a bounded per-consumer memory breakdown to UNABLE_TO_ACQUIRE_MEMORY errors
### What changes were proposed in this pull request?
When a task fails to acquire execution memory, the `UNABLE_TO_ACQUIRE_MEMORY` error
previously reported only the requested and received byte counts:
Unable to acquire 8388608 bytes of memory, got 2097152.
The per-consumer breakdown that `TaskMemoryManager` already computes was written only to the executor logs (`showMemoryUsage()`), so it did not travel with the task failure reason to the driver or the Spark UI.
This PR attaches that per-consumer attribution to the error message itself:
- New `TaskMemoryManager.getMemoryConsumptionBreakdown()` renders a compact, largest-consumer-first breakdown (empty when no consumer holds memory).
- The rendering is factored around a single `snapshotMemoryUsage()` taken under the manager's monitor. On the OOM path, `MemoryConsumer.throwOom()` calls the new `logMemoryUsageAndGetBreakdown()`, which snapshots once and produces **both** the full executor-log dump and the bounded error breakdown from that one snapshot, so the two can never disagree. Snapshotting under the lock (rather than sorting live `getUsed()` values) also avoids tripping `TimSort`'s "Comparison method violates [...]
- The breakdown embedded in the error is bounded by a new internal config `spark.memory.oomErrorConsumerBreakdownLimit` (default 5): the largest consumers are listed individually, the remainder are collapsed into a single summary line that preserves total byte accounting, and `0` omits the breakdown from the error entirely. The full, uncapped breakdown is still written to the executor logs.
- A `consumerBreakdown` parameter is added to the `UNABLE_TO_ACQUIRE_MEMORY` message template and threaded through both construction sites
(`SparkCoreErrors.outOfMemoryError` and
`QueryExecutionErrors.cannotAcquireMemoryForWindowAggregateError`).
### Why are the changes needed?
OOM debugging is a top operational pain point. When a task dies with
`UNABLE_TO_ACQUIRE_MEMORY`, the byte counts alone give no signal about which operator was holding the memory. The information exists, but only in the executor logs, which are often impractical to recover after a failure (rotated / aggregated logs, lost executor) and are not accessible to the driver or to automated / agent-based diagnosis. Surfacing which consumers were competing for memory at the moment of failure, directly in the error that propagates to the driver and the UI, makes [...]
### Does this PR introduce _any_ user-facing change?
Yes. The `UNABLE_TO_ACQUIRE_MEMORY` error message now includes a bounded per-consumer
memory breakdown. For example:
Unable to acquire 8388608 bytes of memory, got 2097152.
Memory used by task 4211 grouped by consumer:
org.apache.spark...UnsafeExternalSorter1a2b: 456.0 MiB
org.apache.spark...BytesToBytesMap3c4d: 12.0 MiB
(37 more consumers): 44.0 MiB
(not attributed to a specific consumer): 3.0 MiB
The breakdown exposes only consumer class names, identity hashes, and aggregate byte counts
-- the same data already written to the executor logs; no keys, values, row data, or paths.
This is a change relative to released Spark versions. The new config
`spark.memory.oomErrorConsumerBreakdownLimit` is internal.
### How was this patch tested?
New unit tests in `TaskMemoryManagerSuite` cover:
- largest-first ordering,
- the cap and the "N more consumers" summary line,
- the limit-0 (omitted) path,
- memory not attributed to any consumer,
- the empty case (no consumer holds memory),
- end-to-end propagation into the thrown `SparkOutOfMemoryError`, and
- `logMemoryUsageAndGetBreakdown()` returning the same breakdown as a standalone render.
Existing `ShuffleExternalSorterSuite`, `WindowSegmentTreeMemorySuite`, and
`SparkThrowableSuite` continue to pass (the last confirms the error-conditions JSON stays well-formed and round-trips).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #57708 from ganeshashree/SPARK-58505.
Lead-authored-by: Ganesha S <ga...@gmail.com>
Co-authored-by: Ganesha S <ga...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../src/main/resources/error/error-conditions.json | 2 +-
.../org/apache/spark/memory/MemoryConsumer.java | 6 +-
.../org/apache/spark/memory/TaskMemoryManager.java | 170 ++++++++++++++++++---
.../org/apache/spark/errors/SparkCoreErrors.scala | 8 +-
.../org/apache/spark/internal/config/package.scala | 15 ++
.../org/apache/spark/memory/MemoryManager.scala | 8 +
.../spark/memory/TaskMemoryManagerSuite.java | 169 ++++++++++++++++++++
.../scala/org/apache/spark/CheckErrorHelper.scala | 5 +-
.../spark/sql/errors/QueryExecutionErrors.scala | 6 +-
.../sql/execution/window/WindowSegmentTree.scala | 2 +-
10 files changed, 360 insertions(+), 31 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json b/common/utils/src/main/resources/error/error-conditions.json
index 7ba37615c364..400fd41b913b 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -8083,7 +8083,7 @@
},
"UNABLE_TO_ACQUIRE_MEMORY" : {
"message" : [
- "Unable to acquire <requestedBytes> bytes of memory, got <receivedBytes>."
+ "Unable to acquire <requestedBytes> bytes of memory, got <receivedBytes>.<consumerBreakdown>"
],
"sqlState" : "53200"
},
diff --git a/core/src/main/java/org/apache/spark/memory/MemoryConsumer.java b/core/src/main/java/org/apache/spark/memory/MemoryConsumer.java
index d2ee6269c134..63ee13756124 100644
--- a/core/src/main/java/org/apache/spark/memory/MemoryConsumer.java
+++ b/core/src/main/java/org/apache/spark/memory/MemoryConsumer.java
@@ -162,7 +162,9 @@ public abstract class MemoryConsumer {
got = page.size();
taskMemoryManager.freePage(page, this);
}
- taskMemoryManager.showMemoryUsage();
- throw SparkCoreErrors.outOfMemoryError(required, got);
+ // Log the full breakdown and attach the bounded one to the error from a single snapshot, so the
+ // executor logs and the driver/UI error message describe the same instant and cannot disagree.
+ String consumerBreakdown = taskMemoryManager.logMemoryUsageAndGetBreakdown();
+ throw SparkCoreErrors.outOfMemoryError(required, got, consumerBreakdown);
}
}
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 099d95c6f9de..cefebcffa2fc 100644
--- a/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
+++ b/core/src/main/java/org/apache/spark/memory/TaskMemoryManager.java
@@ -449,36 +449,162 @@ public class TaskMemoryManager {
}
/**
- * Dump the memory usage of all consumers.
+ * A point-in-time snapshot of this task's execution-memory usage: each consumer that is holding
+ * memory (largest first) paired with its used bytes, plus the bytes not attributable to any
+ * specific consumer. Rendering the executor-log dump and the error-message breakdown from a
+ * single snapshot is what lets the two describe the same instant; see
+ * {@link #logMemoryUsageAndGetBreakdown()}.
*/
- public void showMemoryUsage() {
- logger.info("Memory used in task {}",
- MDC.of(LogKeys.TASK_ATTEMPT_ID, taskAttemptId));
+ private static final class MemoryUsageSnapshot {
+ private final List<Map.Entry<MemoryConsumer, Long>> consumerUsages;
+ private final long memoryNotAccountedFor;
+
+ MemoryUsageSnapshot(
+ List<Map.Entry<MemoryConsumer, Long>> consumerUsages, long memoryNotAccountedFor) {
+ this.consumerUsages = consumerUsages;
+ this.memoryNotAccountedFor = memoryNotAccountedFor;
+ }
+ }
+
+ /**
+ * Snapshot the per-consumer memory usage once, holding the monitor for the whole read.
+ * <p>
+ * {@link MemoryConsumer#getUsed()} reads an {@code AtomicLong} that is not guarded by this
+ * monitor, so callers must not re-read it while sorting or rendering: doing so could observe
+ * changing values and trip {@code TimSort}'s "Comparison method violates its general contract!"
+ * check, masking the OOM we are about to report with an unrelated failure. Consumers are returned
+ * largest first, since the biggest consumers are the most likely culprits of an OOM.
+ */
+ private MemoryUsageSnapshot snapshotMemoryUsage() {
+ List<Map.Entry<MemoryConsumer, Long>> consumerUsages = new ArrayList<>();
+ long memoryAccountedForByConsumers = 0;
+ long memoryNotAccountedFor;
synchronized (this) {
- long memoryAccountedForByConsumers = 0;
- for (MemoryConsumer c: consumers) {
+ for (MemoryConsumer c : consumers) {
long totalMemUsage = c.getUsed();
- memoryAccountedForByConsumers += totalMemUsage;
if (totalMemUsage > 0) {
- logger.info("Acquired by {}: {}",
- MDC.of(LogKeys.MEMORY_CONSUMER, c),
- MDC.of(LogKeys.MEMORY_SIZE, Utils.bytesToString(totalMemUsage)));
+ memoryAccountedForByConsumers += totalMemUsage;
+ consumerUsages.add(new AbstractMap.SimpleEntry<>(c, totalMemUsage));
}
}
- long memoryNotAccountedFor =
+ memoryNotAccountedFor =
memoryManager.getExecutionMemoryUsageForTask(taskAttemptId) - memoryAccountedForByConsumers;
- logger.info(
- "{} bytes of memory were used by task {} but are not associated with specific consumers",
- 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 " +
- "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.MEMORY_SIZE, UnifiedMemoryManager$.MODULE$.getUnmanagedMemoryUsed()));
}
+ consumerUsages.sort(Map.Entry.<MemoryConsumer, Long>comparingByValue().reversed());
+ return new MemoryUsageSnapshot(consumerUsages, memoryNotAccountedFor);
+ }
+
+ /**
+ * Dump the given memory-usage snapshot to the executor logs, one line per consumer (uncapped).
+ */
+ private void logMemoryUsage(MemoryUsageSnapshot snapshot) {
+ logger.info("Memory used in task {}",
+ MDC.of(LogKeys.TASK_ATTEMPT_ID, taskAttemptId));
+ for (Map.Entry<MemoryConsumer, Long> usage : snapshot.consumerUsages) {
+ logger.info("Acquired by {}: {}",
+ MDC.of(LogKeys.MEMORY_CONSUMER, usage.getKey()),
+ MDC.of(LogKeys.MEMORY_SIZE, Utils.bytesToString(usage.getValue())));
+ }
+ logger.info(
+ "{} bytes of memory were used by task {} but are not associated with specific consumers",
+ MDC.of(LogKeys.MEMORY_SIZE, snapshot.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 " +
+ "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.MEMORY_SIZE, UnifiedMemoryManager$.MODULE$.getUnmanagedMemoryUsed()));
+ }
+
+ /**
+ * Render the given snapshot as the compact, bounded breakdown embedded in the
+ * {@code UNABLE_TO_ACQUIRE_MEMORY} error. Returns an empty string when there is nothing to
+ * show -- that is, when the snapshot has neither attributed nor unattributed memory to report
+ * (or when the breakdown is disabled by a limit of 0) -- so callers can append it
+ * unconditionally.
+ */
+ private String renderConsumerBreakdown(MemoryUsageSnapshot snapshot) {
+ // Bound the message that travels to the driver and the UI. The largest consumers -- the most
+ // likely culprits -- are listed individually up to this limit; the rest are collapsed into a
+ // single summary line so total byte accounting is preserved without unbounded noise. The full,
+ // uncapped breakdown is still available in the executor logs via logMemoryUsage(). A limit of
+ // 0 omits the breakdown from the error message entirely.
+ int limit = memoryManager.oomErrorConsumerBreakdownLimit();
+ if (limit == 0) {
+ return "";
+ }
+ List<Map.Entry<MemoryConsumer, Long>> usages = snapshot.consumerUsages;
+ StringBuilder sb = new StringBuilder();
+ int shown = Math.min(limit, usages.size());
+ for (int i = 0; i < shown; i++) {
+ Map.Entry<MemoryConsumer, Long> usage = usages.get(i);
+ sb.append("\n ").append(usage.getKey()).append(": ")
+ .append(Utils.bytesToString(usage.getValue()));
+ }
+ if (usages.size() > shown) {
+ long remainingBytes = 0;
+ for (int i = shown; i < usages.size(); i++) {
+ remainingBytes = Math.addExact(remainingBytes, usages.get(i).getValue());
+ }
+ sb.append("\n (").append(usages.size() - shown).append(" more consumers): ")
+ .append(Utils.bytesToString(remainingBytes));
+ }
+ if (snapshot.memoryNotAccountedFor > 0) {
+ sb.append("\n (not attributed to a specific consumer): ")
+ .append(Utils.bytesToString(snapshot.memoryNotAccountedFor));
+ }
+ if (sb.length() == 0) {
+ return "";
+ }
+ return "\nMemory used by task " + taskAttemptId + " grouped by consumer:" + sb;
+ }
+
+ /**
+ * Dump the memory usage of all consumers to the executor logs.
+ */
+ public void showMemoryUsage() {
+ logMemoryUsage(snapshotMemoryUsage());
+ }
+
+ /**
+ * Build a compact, human-readable breakdown of this task's execution-memory usage grouped by
+ * {@link MemoryConsumer}, with the biggest consumers first, followed by the bytes that are not
+ * attributable to any specific consumer.
+ * <p>
+ * The returned string is meant to be embedded in the {@code UNABLE_TO_ACQUIRE_MEMORY} error so
+ * that the consumers competing for memory at the moment of failure travel with the task failure
+ * reason all the way to the driver and the Spark UI. Returns an empty string when there is
+ * nothing to report -- no consumer is holding memory <i>and</i> there is no unattributed
+ * memory, or the breakdown is disabled by a limit of 0 -- so the caller can append it
+ * unconditionally. Otherwise, a task holding only unattributed memory still gets a breakdown,
+ * consisting of the unattributed line alone.
+ */
+ public String getMemoryConsumptionBreakdown() {
+ // Skip the snapshot entirely when the breakdown is disabled: renderConsumerBreakdown would
+ // discard it anyway. The combined logging path (logMemoryUsageAndGetBreakdown) still needs the
+ // snapshot to write the executor-log dump, so it keeps snapshotting and relies on the
+ // renderer's own limit-0 check.
+ if (memoryManager.oomErrorConsumerBreakdownLimit() == 0) {
+ return "";
+ }
+ return renderConsumerBreakdown(snapshotMemoryUsage());
+ }
+
+ /**
+ * Snapshot this task's memory usage once, write the full (uncapped) breakdown to the executor
+ * logs, and return the bounded breakdown to embed in the {@code UNABLE_TO_ACQUIRE_MEMORY} error.
+ * <p>
+ * Taking a single snapshot for both outputs is what guarantees the log dump and the error message
+ * describe the same instant and cannot disagree; this is the method the OOM path should call
+ * rather than invoking {@link #showMemoryUsage()} and {@link #getMemoryConsumptionBreakdown()}
+ * separately.
+ */
+ public String logMemoryUsageAndGetBreakdown() {
+ MemoryUsageSnapshot snapshot = snapshotMemoryUsage();
+ logMemoryUsage(snapshot);
+ return renderConsumerBreakdown(snapshot);
}
/**
diff --git a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
index a114246c8978..f058a7cd3eda 100644
--- a/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
+++ b/core/src/main/scala/org/apache/spark/errors/SparkCoreErrors.scala
@@ -422,12 +422,16 @@ private[spark] object SparkCoreErrors {
cause = null)
}
- def outOfMemoryError(requestedBytes: Long, receivedBytes: Long): OutOfMemoryError = {
+ def outOfMemoryError(
+ requestedBytes: Long,
+ receivedBytes: Long,
+ consumerBreakdown: String): OutOfMemoryError = {
new SparkOutOfMemoryError(
"UNABLE_TO_ACQUIRE_MEMORY",
Map(
"requestedBytes" -> requestedBytes.toString,
- "receivedBytes" -> receivedBytes.toString).asJava)
+ "receivedBytes" -> receivedBytes.toString,
+ "consumerBreakdown" -> consumerBreakdown).asJava)
}
def failedRenameTempFileError(srcFile: File, dstFile: File): Throwable = {
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 8038307afc65..87a2da236a91 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -503,6 +503,21 @@ package object config {
.doubleConf
.createWithDefault(0.6)
+ private[spark] val MEMORY_OOM_ERROR_CONSUMER_BREAKDOWN_LIMIT =
+ ConfigBuilder("spark.memory.oomErrorConsumerBreakdownLimit")
+ .internal()
+ .doc("The maximum number of memory consumers listed individually in the per-consumer " +
+ "memory breakdown attached to an UNABLE_TO_ACQUIRE_MEMORY error. The largest consumers " +
+ "are listed first; any beyond this limit are collapsed into a single summary line. This " +
+ "bounds the size of the error message that is sent to the driver and shown in the UI. " +
+ "It does not affect the full breakdown written to the executor logs. Set to 0 to omit " +
+ "the breakdown from the error message entirely.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .intConf
+ .checkValue(_ >= 0, "The consumer breakdown limit must not be negative")
+ .createWithDefault(5)
+
private[spark] val UNMANAGED_MEMORY_POLLING_INTERVAL =
ConfigBuilder("spark.memory.unmanagedMemoryPollingInterval")
.doc("Interval for polling unmanaged memory users to track their memory usage. " +
diff --git a/core/src/main/scala/org/apache/spark/memory/MemoryManager.scala b/core/src/main/scala/org/apache/spark/memory/MemoryManager.scala
index 639b82b6080b..dac82cc76357 100644
--- a/core/src/main/scala/org/apache/spark/memory/MemoryManager.scala
+++ b/core/src/main/scala/org/apache/spark/memory/MemoryManager.scala
@@ -62,6 +62,14 @@ private[spark] abstract class MemoryManager(
protected[this] val offHeapStorageMemory =
(maxOffHeapMemory * conf.get(MEMORY_STORAGE_FRACTION)).toLong
+ /**
+ * The maximum number of consumers listed individually in the per-consumer memory breakdown
+ * attached to an UNABLE_TO_ACQUIRE_MEMORY error. Read by [[TaskMemoryManager]]. See
+ * `spark.memory.oomErrorConsumerBreakdownLimit`.
+ */
+ private[memory] val oomErrorConsumerBreakdownLimit: Int =
+ conf.get(MEMORY_OOM_ERROR_CONSUMER_BREAKDOWN_LIMIT)
+
offHeapExecutionMemoryPool.incrementPoolSize(maxOffHeapMemory - offHeapStorageMemory)
offHeapStorageMemoryPool.incrementPoolSize(offHeapStorageMemory)
diff --git a/core/src/test/java/org/apache/spark/memory/TaskMemoryManagerSuite.java b/core/src/test/java/org/apache/spark/memory/TaskMemoryManagerSuite.java
index 6c408c0c928b..44e17d040ea3 100644
--- a/core/src/test/java/org/apache/spark/memory/TaskMemoryManagerSuite.java
+++ b/core/src/test/java/org/apache/spark/memory/TaskMemoryManagerSuite.java
@@ -151,6 +151,26 @@ public class TaskMemoryManagerSuite {
}
}
+ private static final class NonSpillingConsumer extends MemoryConsumer {
+ NonSpillingConsumer(TaskMemoryManager manager) {
+ super(manager, 1024L, MemoryMode.ON_HEAP);
+ }
+
+ void use(long size) {
+ used.getAndAdd(taskMemoryManager.acquireExecutionMemory(size, this));
+ }
+
+ void free(long size) {
+ used.getAndAdd(-size);
+ taskMemoryManager.releaseExecutionMemory(size, this);
+ }
+
+ @Override
+ public long spill(long size, MemoryConsumer trigger) {
+ return 0;
+ }
+ }
+
private static final class NonSpillingAllocatingConsumer extends TestMemoryConsumer {
private int spillAttempts;
private MemoryBlock nestedPage;
@@ -841,6 +861,155 @@ public class TaskMemoryManagerSuite {
Assertions.assertEquals(80, c1.getUsed()); // not spilled
}
+ @Test
+ public void memoryConsumptionBreakdownIsEmptyWhenThereIsNoMemoryToReport() {
+ final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
+ memoryManager.limit(100);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
+ // The breakdown is empty only when the task holds neither attributed nor unattributed memory.
+ // A consumer that has never acquired memory must not appear in the breakdown.
+ new TestMemoryConsumer(manager);
+ Assertions.assertEquals(0, memoryManager.getExecutionMemoryUsageForTask(0));
+ Assertions.assertEquals("", manager.getMemoryConsumptionBreakdown());
+ }
+
+ @Test
+ public void memoryConsumptionBreakdownListsConsumersLargestFirst() {
+ final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
+ memoryManager.limit(100);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
+
+ TestMemoryConsumer small = new TestMemoryConsumer(manager);
+ TestMemoryConsumer large = new TestMemoryConsumer(manager);
+ small.use(20);
+ large.use(60);
+
+ String breakdown = manager.getMemoryConsumptionBreakdown();
+ Assertions.assertTrue(breakdown.contains("grouped by consumer"), breakdown);
+ // The larger consumer is listed before the smaller one so the likely culprit surfaces first.
+ int largeIdx = breakdown.indexOf(large.toString());
+ int smallIdx = breakdown.indexOf(small.toString());
+ Assertions.assertTrue(largeIdx >= 0 && smallIdx >= 0, breakdown);
+ Assertions.assertTrue(largeIdx < smallIdx, breakdown);
+
+ small.free(20);
+ large.free(60);
+ Assertions.assertEquals(0, manager.cleanUpAllAllocatedMemory());
+ }
+
+ @Test
+ public void memoryConsumptionBreakdownCapsConsumersAndSummarizesTheRest() {
+ final SparkConf conf = new SparkConf()
+ .set(package$.MODULE$.MEMORY_OOM_ERROR_CONSUMER_BREAKDOWN_LIMIT(), 2);
+ final TestMemoryManager memoryManager = new TestMemoryManager(conf);
+ memoryManager.limit(1000);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
+
+ TestMemoryConsumer c1 = new TestMemoryConsumer(manager);
+ TestMemoryConsumer c2 = new TestMemoryConsumer(manager);
+ TestMemoryConsumer c3 = new TestMemoryConsumer(manager);
+ TestMemoryConsumer c4 = new TestMemoryConsumer(manager);
+ c1.use(40);
+ c2.use(30);
+ c3.use(20);
+ c4.use(10);
+
+ String breakdown = manager.getMemoryConsumptionBreakdown();
+ // Only the two largest consumers are listed individually.
+ Assertions.assertTrue(breakdown.contains(c1.toString()), breakdown);
+ Assertions.assertTrue(breakdown.contains(c2.toString()), breakdown);
+ Assertions.assertFalse(breakdown.contains(c3.toString()), breakdown);
+ Assertions.assertFalse(breakdown.contains(c4.toString()), breakdown);
+ // The remaining two are collapsed into a summary line covering their combined 30 bytes.
+ Assertions.assertTrue(breakdown.contains("(2 more consumers): 30.0 B"), breakdown);
+
+ c1.free(40);
+ c2.free(30);
+ c3.free(20);
+ c4.free(10);
+ Assertions.assertEquals(0, manager.cleanUpAllAllocatedMemory());
+ }
+
+ @Test
+ public void memoryConsumptionBreakdownIsOmittedWhenLimitIsZero() {
+ final SparkConf conf = new SparkConf()
+ .set(package$.MODULE$.MEMORY_OOM_ERROR_CONSUMER_BREAKDOWN_LIMIT(), 0);
+ final TestMemoryManager memoryManager = new TestMemoryManager(conf);
+ memoryManager.limit(100);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
+ final TestMemoryConsumer c = new TestMemoryConsumer(manager);
+ c.use(50);
+
+ Assertions.assertEquals("", manager.getMemoryConsumptionBreakdown());
+
+ c.free(50);
+ Assertions.assertEquals(0, manager.cleanUpAllAllocatedMemory());
+ }
+
+ @Test
+ public void memoryConsumptionBreakdownReportsMemoryNotAttributedToAConsumer() {
+ final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
+ memoryManager.limit(4096);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
+ // Acquire task memory that is not tracked by any registered consumer's getUsed(). The
+ // NonSpillingConsumer intentionally does not increment its `used` counter, so from the
+ // manager's perspective this memory is used by the task but unattributed.
+ final NonSpillingConsumer c = new NonSpillingConsumer(manager);
+ manager.acquireExecutionMemory(2048, c);
+
+ String breakdown = manager.getMemoryConsumptionBreakdown();
+ Assertions.assertTrue(
+ breakdown.contains("(not attributed to a specific consumer)"), breakdown);
+ // The consumer never bumped its own counter, so it must not appear as an attributed line.
+ Assertions.assertFalse(breakdown.contains(c.toString() + ": "), breakdown);
+
+ manager.releaseExecutionMemory(2048, c);
+ Assertions.assertEquals("", manager.getMemoryConsumptionBreakdown());
+ }
+
+ @Test
+ public void outOfMemoryErrorCarriesConsumerBreakdown() {
+ final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
+ memoryManager.limit(1024);
+ final TestAllocator allocator = new TestAllocator(Integer.MAX_VALUE);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0, allocator);
+ // A non-spilling consumer holds all the memory so the page allocation below cannot be
+ // satisfied and MemoryConsumer#throwOom fires.
+ final NonSpillingConsumer hog = new NonSpillingConsumer(manager);
+ hog.use(1024);
+ final PageAllocatingConsumer requestingConsumer = new PageAllocatingConsumer(manager, 4096);
+
+ SparkOutOfMemoryError e = Assertions.assertThrows(
+ SparkOutOfMemoryError.class, () -> requestingConsumer.allocate(4096));
+ Assertions.assertEquals("UNABLE_TO_ACQUIRE_MEMORY", e.getCondition());
+ String breakdown = e.getMessageParameters().get("consumerBreakdown");
+ Assertions.assertNotNull(breakdown);
+ // The breakdown captured at failure time names the consumer that was hogging memory.
+ Assertions.assertTrue(breakdown.contains(hog.toString()), breakdown);
+ Assertions.assertTrue(e.getMessage().contains("grouped by consumer"), e.getMessage());
+
+ hog.free(1024);
+ Assertions.assertEquals(0, manager.cleanUpAllAllocatedMemory());
+ }
+
+ @Test
+ public void logMemoryUsageAndGetBreakdownReturnsSameBreakdownAsRenderer() {
+ final TestMemoryManager memoryManager = new TestMemoryManager(new SparkConf());
+ memoryManager.limit(100);
+ final TaskMemoryManager manager = new TaskMemoryManager(memoryManager, 0);
+ TestMemoryConsumer c = new TestMemoryConsumer(manager);
+ c.use(50);
+
+ // The combined OOM-path method logs and returns the breakdown from a single snapshot; on a
+ // quiescent manager the returned breakdown matches a standalone render of the same state.
+ String combined = manager.logMemoryUsageAndGetBreakdown();
+ Assertions.assertEquals(manager.getMemoryConsumptionBreakdown(), combined);
+ Assertions.assertTrue(combined.contains(c.toString()), combined);
+
+ c.free(50);
+ Assertions.assertEquals(0, manager.cleanUpAllAllocatedMemory());
+ }
+
@Test
public void offHeapConfigurationBackwardsCompatibility() {
// Tests backwards-compatibility with the old `spark.unsafe.offHeap` configuration, which
diff --git a/core/src/test/scala/org/apache/spark/CheckErrorHelper.scala b/core/src/test/scala/org/apache/spark/CheckErrorHelper.scala
index d01600bb439f..d0fbb8ead436 100644
--- a/core/src/test/scala/org/apache/spark/CheckErrorHelper.scala
+++ b/core/src/test/scala/org/apache/spark/CheckErrorHelper.scala
@@ -67,7 +67,10 @@ trait CheckErrorHelper { self: Suite =>
* Test suites may override this to add or change ignorable parameters per condition.
*/
protected def checkErrorIgnorableParameters: Map[String, Set[String]] = Map(
- "TABLE_OR_VIEW_NOT_FOUND" -> Set("searchPath")
+ "TABLE_OR_VIEW_NOT_FOUND" -> Set("searchPath"),
+ // The per-consumer memory breakdown is a best-effort diagnostic whose content (live memory
+ // sizes and consumer identities) is inherently non-deterministic, so tests need not pin it.
+ "UNABLE_TO_ACQUIRE_MEMORY" -> Set("consumerBreakdown")
)
/**
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index 4be4f271f1cf..7d00aa292a3c 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -1223,12 +1223,14 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
def cannotAcquireMemoryForWindowAggregateError(
requestedBytes: Long,
- receivedBytes: Long): SparkOutOfMemoryError = {
+ receivedBytes: Long,
+ consumerBreakdown: String): SparkOutOfMemoryError = {
new SparkOutOfMemoryError(
"UNABLE_TO_ACQUIRE_MEMORY",
java.util.Map.of(
"requestedBytes", requestedBytes.toString,
- "receivedBytes", receivedBytes.toString))
+ "receivedBytes", receivedBytes.toString,
+ "consumerBreakdown", consumerBreakdown))
}
def rowLargerThan256MUnsupportedError(): SparkUnsupportedOperationException = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
index cdc6556f1862..e40267bd7448 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/window/WindowSegmentTree.scala
@@ -434,7 +434,7 @@ private[window] class WindowSegmentTree(
if (!evictEldest() || !acquireBlockMemory()) {
// scalastyle:off throwerror
throw QueryExecutionErrors.cannotAcquireMemoryForWindowAggregateError(
- blockBytes, 0L)
+ blockBytes, 0L, taskMemoryManager.getMemoryConsumptionBreakdown())
// scalastyle:on throwerror
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org