[SPARK-59776][CORE][SQL] Notify ExecutorMonitor when SQL shuffle cleanup removes shuffle files

[saurabhdamle11]

What changes were proposed in this pull request?

When SQL shuffle dependency file cleanup is enabled (spark.sql.classic.shuffleDependency.fileCleanup.enabled, or the connect / thriftserver variants), SQLExecution.cleanupShuffleDependencies removes a query's shuffle files by calling ShuffleDriverComponents.removeShuffle directly. It doesn't go through ContextCleaner, so CleanerListeners are never told the shuffle is gone.

This PR:

• Adds CleanerListener.shuffleFilesRemoved(shuffleId), a no-op by default.
• Adds ContextCleaner.notifyShuffleFilesRemoved(shuffleId), which calls it on all attached listeners.
• Calls it from SQLExecution after a successful RemoveShuffleFiles cleanup.
• Overrides it in the dynamic allocation ExecutorMonitor, so that executors stop tracking the removed shuffle.

A new callback is used instead of reusing shuffleCleaned. shuffleCleaned means the shuffle was fully cleaned, including unregistering its map outputs from the MapOutputTracker. This cleanup path deliberately doesn't unregister them, so that stage retries can still be triggered, and existing shuffleCleaned listeners rely on that meaning.

Why are the changes needed?

With spark.dynamicAllocation.shuffleTracking.enabled, ExecutorMonitor keeps an executor alive while it holds shuffle data that may still be used. It stops tracking a shuffle only when ContextCleaner reports it cleaned, which happens after the ShuffleDependency is garbage collected on the driver.

With SQL shuffle file cleanup enabled, the shuffle files are deleted right after the query finishes, but ExecutorMonitor isn't told. It keeps executors pinned to shuffles that no longer exist until the ShuffleDependency is eventually garbage collected, which may never happen while the DataFrame is still referenced. With the default spark.dynamicAllocation.shuffleTracking.timeout (infinite), those executors are never released even though they hold no shuffle data.

Does this PR introduce any user-facing change?

Yes. When SQL shuffle dependency file cleanup and dynamic allocation shuffle tracking are both enabled, executors whose shuffle files were removed by the cleanup are now released after spark.dynamicAllocation.executorIdleTimeout. Before this change they stayed until the shuffle was garbage collected on the driver, or until spark.dynamicAllocation.shuffleTracking.timeout. Behavior with the default configuration (file cleanup disabled) is unchanged.

How was this patch tested?

New unit tests:

• ExecutorMonitorSuite: "SPARK-59776: executor with removed shuffle files times out at the idle deadline". It checks that after shuffleFilesRemoved, an executor that only held that shuffle times out at the idle timeout instead of the shuffle tracking timeout.
• SQLExecutionSuite: "SPARK-59776: RemoveShuffleFiles cleanup notifies cleaner listeners". It runs a shuffle join with file cleanup enabled and checks that shuffleFilesRemoved fires for every shuffle in the plan and that shuffleCleaned doesn't.

All tests passed on GitHub Actions: https://github.com/saurabhdamle11/spark/actions/runs/36168154179

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Opus 5.5

Title: [SPARK-59776][CORE][SQL] Release shuffle-tracked executors when SQL shuffle file cleanup removes shuffle files