[SPARK-58505][CORE] Attach a bounded per-consumer memory breakdown to UNABLE_TO_ACQUIRE_MEMORY errors

[ganeshashree]

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 its general contract!" check on the failure path.
• 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 these failures far easier to diagnose.

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...UnsafeExternalSorter@1a2b: 456.0 MiB
  org.apache.spark...BytesToBytesMap@3c4d: 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)

[uros-b]

Left just one comment, otherwise looks good - thank you @ganeshashree!

[cloud-fan]

0 blocking, 0 non-blocking, 1 nit.
The implementation is coherent and well covered; one inaccurate Javadoc sentence should be corrected.

Nits: 1 minor item (see inline comments).

Verification

I traced both OOM construction paths through the shared renderer and checked the config's default, non-negative validation, zero sentinel, bounded summary, and error-parameter propagation. The focused tests cover ordering, truncation, unattributed memory, omission at zero, and the generic allocation failure path.

[cloud-fan]

@ganeshashree can you make the CI happy?

[ganeshashree]

@ganeshashree can you make the CI happy?

Created this PR to self-heal the PR build check when the notify workflow fails to create it.