diff --git a/core/src/main/scala/org/apache/spark/BlockTTLCleaner.scala b/core/src/main/scala/org/apache/spark/BlockTTLCleaner.scala
new file mode 100644
index 0000000000000..5a48634d60b8f
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/BlockTTLCleaner.scala
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark
+
+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.{Logging, LogKey}
+import org.apache.spark.internal.LogKeys.{COMPONENT, TIMEOUT}
+import org.apache.spark.util.ThreadUtils
+
+/**
+ * Periodic TTL sweep over an id -> last-access-time map, on a daemon thread of its own: reaps every
+ * id idle for longer than `ttlMillis`, then sleeps until the next one can expire. Shared by the
+ * RDD-cache TTL cleaner (`BlockManagerMasterEndpoint`) and the shuffle TTL cleaner
+ * (`MapOutputTrackerMaster`), which differ only in the map, the `shouldReap` veto and the `reap`
+ * action.
+ *
+ * `accessTimes` must be concurrent: this thread iterates and removes from it while others put.
+ * `shouldReap` returning false leaves the id tracked for the next pass (e.g. a shuffle that has
+ * not produced output yet has nothing to reclaim).
+ */
+private[spark] class BlockTTLCleaner(
+    name: String,
+    idKey: LogKey,
+    ttlMillis: Long,
+    accessTimes: ConcurrentHashMap[Int, Long],
+    shouldReap: Int => Boolean,
+    reap: Int => Unit) extends Runnable with Logging {
+
+  import BlockTTLCleaner.StopTimeoutSeconds
+
+  private val pool = ThreadUtils.newDaemonSingleThreadExecutor(s"$name-ttl-cleaner")
+
+  // Not just the interrupt from shutdownNow: a reap can swallow the InterruptedException on its way
+  // out (ContextCleaner.doCleanupRDD catches Exception), which also clears the interrupt flag, and
+  // then nothing would ever stop the sweep.
+  @volatile private var stopped = false
+
+  def start(): Unit = pool.execute(this)
+
+  def stop(): Unit = {
+    stopped = true
+    pool.shutdownNow()
+    // Wait for a reap already in flight, briefly. SparkContext.stop stops these cleaners before the
+    // ContextCleaner, the listener bus and the shuffle driver components precisely so a reap cannot
+    // use them half torn down; without joining, that ordering only narrows the window instead of
+    // closing it. Bounded because a reap can be blocked on an RPC to a dead executor: shutdownNow
+    // has already interrupted it, and a daemon thread outliving this call is better than a
+    // shutdown that hangs.
+    if (!pool.awaitTermination(StopTimeoutSeconds, TimeUnit.SECONDS)) {
+      logWarning(log"The ${MDC(COMPONENT, name)} TTL cleaner did not stop within " +
+        log"${MDC(TIMEOUT, StopTimeoutSeconds)}s; continuing shutdown with a reap still in flight")
+    }
+  }
+
+  /**
+   * One pass over `accessTimes`: reap whatever has expired, and return how long to wait before the
+   * next pass. Separate from `run` so a test can drive a single sweep rather than wait out a TTL --
+   * the configured minimum is 10 minutes, so waiting is not an option for a test.
+   */
+  private[spark] def sweep(): Long = {
+    val now = System.currentTimeMillis()
+    // Oldest live atime, so we can sleep until the next possible expiry.
+    var oldestLive = now
+    val expired = List.newBuilder[(Int, Long)]
+    accessTimes.forEach { (id, atime) =>
+      if (atime >= now - ttlMillis) oldestLive = math.min(oldestLive, atime)
+      else expired += ((id, atime))
+    }
+    // Reap outside the iteration: a reap blocks on an RPC, and comparing the rest of the map
+    // against a `now` that went stale meanwhile would defer them a whole sweep.
+    expired.result().foreach { case (id, atime) =>
+      // remove(k, v) fails if the atime moved since we read it: the id is back in use.
+      if (shouldReap(id) && accessTimes.remove(id, atime)) {
+        var reaped = false
+        try {
+          reap(id)
+          reaped = true
+        } catch {
+          // Warn, not debug: reclaiming space is this loop's whole job.
+          case NonFatal(e) =>
+            logWarning(log"Error reaping ${MDC(idKey, id)} in the " +
+              log"${MDC(COMPONENT, name)} TTL cleaner", e)
+        } finally {
+          if (!reaped) {
+            // Start the id's clock again rather than leaving it untracked. A reap can fail
+            // part-way -- a blocking shuffle removal times out waiting on an executor, say --
+            // and dropping the atime here would mean nothing ever revisits the id, so whatever
+            // it still holds on disk would leak for the life of the driver. A fresh timestamp
+            // retries one TTL from now instead of every sweep, so a reap that always fails
+            // costs one warning per TTL. putIfAbsent, because the id may have come back into
+            // use while the reap ran, and that atime is the more recent truth.
+            accessTimes.putIfAbsent(id, System.currentTimeMillis())
+          }
+        }
+      }
+    }
+    // Wait until the oldest live entry can expire, but never spin: with a live entry always near
+    // expiry the floor is what stops this from rescanning the whole map continuously.
+    val floor = math.max(ttlMillis / 10, 100)
+    math.max(oldestLive + ttlMillis - System.currentTimeMillis(), floor)
+  }
+
+  override def run(): Unit = {
+    try {
+      while (!stopped && !Thread.currentThread().isInterrupted) {
+        val sleepMillis = sweep()
+        // Re-check `stopped`: a reap that swallowed the interrupt cleared the flag too, and
+        // sleeping out a whole TTL would keep the stopped SparkContext alive via the reap closure.
+        if (!stopped) {
+          Thread.sleep(sleepMillis)
+        }
+      }
+    } catch {
+      case _: InterruptedException => // Shutdown; fall through and exit.
+      // Nothing restarts this thread, so say so rather than stopping reaping in silence.
+      case NonFatal(e) =>
+        logError(log"The ${MDC(COMPONENT, name)} TTL cleaner died; nothing more will be reaped", e)
+    }
+    logInfo(log"${MDC(COMPONENT, name)} TTL cleaner exiting.")
+  }
+}
+
+private[spark] object BlockTTLCleaner {
+  // How long stop() waits for an in-flight reap before giving up on it.
+  val StopTimeoutSeconds = 2L
+}
diff --git a/core/src/main/scala/org/apache/spark/ContextCleaner.scala b/core/src/main/scala/org/apache/spark/ContextCleaner.scala
index 6c1b49157cc01..67cdd0b1cd825 100644
--- a/core/src/main/scala/org/apache/spark/ContextCleaner.scala
+++ b/core/src/main/scala/org/apache/spark/ContextCleaner.scala
@@ -124,6 +124,21 @@ private[spark] class ContextCleaner(
     listeners.add(listener)
   }
 
+  /**
+   * Notify listeners of a shuffle reclaimed outside the reference-tracking path, i.e. by the
+   * shuffle TTL cleaner. Needed so that `ExecutorMonitor` sees the shuffle go away and dynamic
+   * allocation can release an executor that only held it.
+   */
+  private[spark] def notifyShuffleCleaned(shuffleId: Int): Unit = {
+    // Contained the same way doCleanupShuffle contains its own fan-out. This runs on the shuffle
+    // TTL cleaner's thread, at the end of a reap that has already deleted the files and
+    // unregistered the shuffle, so letting a listener's exception escape would only report a
+    // completed reap as failed.
+    Utils.tryLogNonFatalError {
+      listeners.asScala.foreach(_.shuffleCleaned(shuffleId))
+    }
+  }
+
   /** Start the cleaner. */
   def start(): Unit = {
     cleaningThread.setDaemon(true)
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index 933da94772725..c40bcad712830 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -801,6 +801,35 @@ private[spark] class MapOutputTrackerMaster(
     private[spark] val isLocal: Boolean)
   extends MapOutputTracker(conf) with ShuffleOutputTrackerMaster {
 
+  // Last access time per shuffle, for the shuffle TTL. Overwriting a "close" timestamp is fine, but
+  // the map-output dispatcher, the DAGScheduler and the TTL cleaner all write it, so it has to be
+  // a concurrent map.
+  private[spark] val shuffleAccessTime = new ConcurrentHashMap[Int, Long]
+
+  // Reclaims a shuffle's on-disk blocks via the normal ShuffleDriverComponents.removeShuffle path.
+  // Wired by SparkContext, which owns those components; this tracker is built earlier, in SparkEnv.
+  // None until then, and the cleaner refuses to reap while it is None: reaping would drop the
+  // driver's shuffle state while reclaiming no executor disk at all, leaking those files for the
+  // life of the application. Mirrors how `rddReapable` defaults to refusing everything.
+  @volatile private[spark] var shuffleFileRemover: Option[Int => Unit] = None
+
+  // Tells the ContextCleaner's listeners a shuffle is gone, so shuffle-tracking dynamic allocation
+  // can release an executor that only held it. Separate from `shuffleFileRemover` so the reap can
+  // run it after `unregisterShuffle` rather than before; best-effort, so it does not gate a reap.
+  @volatile private[spark] var shuffleCleanedNotifier: Option[Int => Unit] = None
+
+  // Read once: updateShuffleAtime is on the hot path of serving map output requests. The config is
+  // in seconds; the cleaner works in millis, like the access times it compares against.
+  private val shuffleTtl: Option[Long] =
+    conf.get(CLEANER_TTL_SHUFFLE).map(TimeUnit.SECONDS.toMillis)
+
+  /** Records that a shuffle was used, for the shuffle TTL. A no-op when that TTL is unset. */
+  private def updateShuffleAtime(shuffleId: Int): Unit = {
+    if (shuffleTtl.isDefined) {
+      shuffleAccessTime.put(shuffleId, System.currentTimeMillis())
+    }
+  }
+
   // The size at which we use Broadcast to send the map output statuses to the executors
   private val minSizeForBroadcast = conf.get(SHUFFLE_MAPOUTPUT_MIN_SIZE_FOR_BROADCAST).toInt
 
@@ -865,6 +894,45 @@ private[spark] class MapOutputTrackerMaster(
     throw new IllegalArgumentException(logEntry.message)
   }
 
+  // Started at the end of construction, after the fail-fast check above, so a failed construction
+  // can't orphan the thread.
+  //
+  // A registered shuffle that has produced no output yet (e.g. a map stage waiting on parents) has
+  // nothing to reclaim, so it stays tracked. Anything else is reaped, including an id with no
+  // ShuffleStatus at all -- otherwise its atime would be vetoed forever and never dropped. Nothing
+  // is reaped at all until SparkContext wires `shuffleFileRemover`; see there.
+  private[spark] val ttlCleaner: Option[BlockTTLCleaner] = shuffleTtl.map { ttl =>
+    new BlockTTLCleaner(
+      name = "shuffle",
+      idKey = SHUFFLE_ID,
+      ttlMillis = ttl,
+      accessTimes = shuffleAccessTime,
+      shouldReap = shuffleId =>
+        shuffleFileRemover.isDefined &&
+          !shuffleStatuses.get(shuffleId).exists(_.numAvailableMapOutputs == 0),
+      reap = shuffleId => {
+        // Bump the epoch before deleting anything. This does not protect a reduce task that is
+        // already running -- an executor only picks up a new epoch when its next task is launched
+        // (Executor.updateEpoch), so a task holding cached statuses still fetches the deleted files
+        // and fails. It does keep tasks launched from here on from being handed locations that are
+        // about to disappear. `docs/configuration.md` spells out the job-failure tail.
+        if (shuffleStatuses.contains(shuffleId)) {
+          incrementEpoch()
+        }
+        // Delete before dropping the map output, not after: removeShuffle reads the live
+        // MapStatuses and push-merger locations to find what the external shuffle service must
+        // delete. Same ordering, same reason, as ContextCleaner.doCleanupShuffle.
+        shuffleFileRemover.foreach(_(shuffleId))
+        // Drops the whole ShuffleStatus. The async RemoveShuffle broadcast would too; doing it here
+        // makes the outcome independent of when that lands.
+        unregisterShuffle(shuffleId)
+        // Last, so a CleanerListener never observes shuffleCleaned while the shuffle is still
+        // registered -- the order doCleanupShuffle establishes.
+        shuffleCleanedNotifier.foreach(_(shuffleId))
+      })
+  }
+  ttlCleaner.foreach(_.start())
+
   def post(message: MapOutputTrackerMasterMessage): Unit = {
     mapOutputTrackerMasterMessages.offer(message)
   }
@@ -879,6 +947,7 @@ private[spark] class MapOutputTrackerMaster(
       val shuffleStatus = shuffleStatuses.get(shuffleId).head
       logDebug(s"Handling request to send ${if (needMergeOutput) "map/merge" else "map"}" +
         s" output locations for shuffle $shuffleId to $hostPort")
+      updateShuffleAtime(shuffleId)
       if (needMergeOutput) {
         context.reply(
           shuffleStatus.
@@ -930,6 +999,7 @@ private[spark] class MapOutputTrackerMaster(
   }
 
   def registerShuffle(shuffleId: Int, numMaps: Int, numReduces: Int): Unit = {
+    updateShuffleAtime(shuffleId)
     if (pushBasedShuffleEnabled) {
       if (shuffleStatuses.put(shuffleId,
         new ShuffleStatus(numMaps, numReduces, bufferRacingMigrations)).isDefined) {
@@ -950,6 +1020,10 @@ private[spark] class MapOutputTrackerMaster(
   def updateMapOutput(shuffleId: Int, mapId: Long, bmAddress: BlockManagerId): Unit = {
     shuffleStatuses.get(shuffleId) match {
       case Some(shuffleStatus) =>
+        // A migration can restore output that unregisterAllMapAndMergeOutput had moved into
+        // mapStatusesDeleted, and that path dropped the atime. Without re-stamping here the shuffle
+        // would have live outputs and files but no atime, so the TTL would never reclaim it.
+        updateShuffleAtime(shuffleId)
         shuffleStatus.updateMapOutput(mapId, bmAddress)
       case None if shuffleMigrationEnabled =>
         logWarning(log"Asked to update map output for unknown shuffle " +
@@ -967,6 +1041,9 @@ private[spark] class MapOutputTrackerMaster(
   }
 
   def registerMapOutput(shuffleId: Int, mapIndex: Int, status: MapStatus): Boolean = {
+    // Producing output counts as a use, so a map stage running longer than the TTL is not reaped
+    // mid-production (registerShuffle only stamps the atime once, at stage submission).
+    updateShuffleAtime(shuffleId)
     getShuffleStatusOrError(shuffleId, "registerMapOutput").addMapOutput(mapIndex, status)
   }
 
@@ -978,6 +1055,7 @@ private[spark] class MapOutputTrackerMaster(
 
   /** Unregister all map and merge output information of the given shuffle. */
   def unregisterAllMapAndMergeOutput(shuffleId: Int): Unit = {
+    shuffleAccessTime.remove(shuffleId)
     val shuffleStatus = getShuffleStatusOrError(shuffleId, "unregisterAllMapAndMergeOutput")
     shuffleStatus.removeOutputsByFilter(x => true)
     shuffleStatus.removeMergeResultsByFilter(x => true)
@@ -1035,6 +1113,9 @@ private[spark] class MapOutputTrackerMaster(
 
   /** Unregister shuffle data */
   override def unregisterShuffle(shuffleId: Int): Unit = {
+    // Stop tracking, so the cleaner can't later reap a shuffle the ContextCleaner already collected
+    // (which would throw ShuffleStatusNotFoundException).
+    shuffleAccessTime.remove(shuffleId)
     shuffleStatuses.remove(shuffleId).foreach { shuffleStatus =>
       shuffleStatus.invalidateSerializedMapOutputStatusCache()
       shuffleStatus.invalidateSerializedMergeOutputStatusCache()
@@ -1346,6 +1427,9 @@ private[spark] class MapOutputTrackerMaster(
       startPartition: Int,
       endPartition: Int): MapSizesByExecutorId = {
     logDebug(s"Fetching outputs for shuffle $shuffleId")
+    // The local-mode read path: no RPC round trip, so this is the only place a local-mode reduce
+    // task's use of the shuffle can be seen (handleStatusMessage covers the RPC one).
+    updateShuffleAtime(shuffleId)
     shuffleStatuses.get(shuffleId) match {
       case Some(shuffleStatus) =>
         shuffleStatus.withMapStatuses { statuses =>
@@ -1377,11 +1461,28 @@ private[spark] class MapOutputTrackerMaster(
     Seq.empty.iterator
   }
 
-  // This method is only called in local-mode.
+  // Called in local-mode, and by BlockManagerMaster.removeShuffle to snapshot the merger locations
+  // for a shuffle it is about to remove. No atime update, because of that second caller: stamping
+  // here would resurrect the atime of a shuffle that is going away.
   override def getShufflePushMergerLocations(shuffleId: Int): Seq[BlockManagerId] = {
     shuffleStatuses.get(shuffleId).map(_.getShufflePushMergerLocations).getOrElse(Seq.empty)
   }
 
+  /**
+   * The location and map id of every available map output of a shuffle. Used by the shuffle removal
+   * path to work out which blocks an external shuffle service still holds on behalf of executors
+   * that have since gone away. Reading this takes the shuffle's `ShuffleStatus` lock, so it is
+   * snapshotted on the caller's thread -- see `BlockManagerMaster.removeShuffle`. No atime update,
+   * for the same reason as `getShufflePushMergerLocations` above.
+   */
+  private[spark] def getMapOutputLocations(shuffleId: Int): Seq[(BlockManagerId, Long)] = {
+    shuffleStatuses.get(shuffleId).map { shuffleStatus =>
+      shuffleStatus.withMapStatuses { statuses =>
+        statuses.filter(_ != null).map(status => (status.location, status.mapId)).toSeq
+      }
+    }.getOrElse(Seq.empty)
+  }
+
   override def getStaleMapIndexes(shuffleId: Int): Set[Int] = {
     shuffleStatuses.get(shuffleId).map(_.getStaleMapIndexes).getOrElse(Set.empty)
   }
@@ -1389,6 +1490,7 @@ private[spark] class MapOutputTrackerMaster(
   override def stop(): Unit = {
     mapOutputTrackerMasterMessages.offer(PoisonPill)
     threadpool.shutdown()
+    ttlCleaner.foreach(_.stop())
     try {
       sendTracker(StopMapOutputTracker)
     } catch {
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 438c036a359af..03e42c8bb3c0a 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -232,7 +232,9 @@ class SparkContext(config: SparkConf) extends Logging {
   private var _eventLogger: Option[EventLoggingListener] = None
   private var _driverLogger: Option[DriverLogger] = None
   private var _executorAllocationManager: Option[ExecutorAllocationManager] = None
-  private var _cleaner: Option[ContextCleaner] = None
+  // @volatile because the TTL cleaner threads read it from their reap closures, and it is assigned
+  // after those closures are published.
+  @volatile private var _cleaner: Option[ContextCleaner] = None
   private var _listenerBusStarted: Boolean = false
   private var _jars: Seq[String] = _
   private var _files: Seq[String] = _
@@ -321,6 +323,13 @@ class SparkContext(config: SparkConf) extends Logging {
     val map: ConcurrentMap[Int, RDD[_]] = new MapMaker().weakValues().makeMap[Int, RDD[_]]()
     map.asScala
   }
+
+  // The RDD TTL cleaner must never reap a locally checkpointed RDD: local checkpointing truncates
+  // lineage, so its cache blocks are the only copy of the data. Asked of the RDD itself rather than
+  // tracked alongside; a reaped id is gone from persistentRdds, which is the same answer.
+  private[spark] def isLocallyCheckpointedRdd(rddId: Int): Boolean =
+    persistentRdds.get(rddId).exists(_.isLocallyCheckpointed)
+
   def statusTracker: SparkStatusTracker = _statusTracker
 
   private[spark] def progressBar: Option[ConsoleProgressBar] = _progressBar
@@ -661,6 +670,43 @@ class SparkContext(config: SparkConf) extends Logging {
       _conf.set(ShuffleDataIOUtils.SHUFFLE_SPARK_CONF_PREFIX + k, v)
     }
 
+    // Wire up the TTL cleaners: they live in SparkEnv components built before the pieces they need
+    // (ShuffleDriverComponents, the ContextCleaner and this SparkContext) exist.
+    _env.mapOutputTracker match {
+      case mapOutputTrackerMaster: MapOutputTrackerMaster =>
+        // Read once, not per reap, like the other TTL configs.
+        val blockingShuffleRemoval = _conf.get(CLEANER_REFERENCE_TRACKING_BLOCKING_SHUFFLE)
+        // Together these mirror the tail of ContextCleaner.doCleanupShuffle: delete the files, and
+        // (after the tracker has unregistered the shuffle) notify the listeners so shuffle-tracking
+        // dynamic allocation can release an executor that only held this shuffle.
+        mapOutputTrackerMaster.shuffleFileRemover = Some { shuffleId =>
+          _shuffleDriverComponents.removeShuffle(shuffleId, blockingShuffleRemoval)
+        }
+        mapOutputTrackerMaster.shuffleCleanedNotifier = Some { shuffleId =>
+          _cleaner.foreach(_.notifyShuffleCleaned(shuffleId))
+        }
+      case _ =>
+    }
+    _env.blockManagerMasterEndpoint.foreach { endpoint =>
+      endpoint.rddReapable = rddId => !isLocallyCheckpointedRdd(rddId)
+      // Reap through the same entry point the GC-driven cleanup uses; unpersistRDD is the fallback
+      // when reference tracking is off. Both free the blocks without resetting the RDD's storage
+      // level (the cleaner only has an id), so a later action re-caches it -- but both also drop
+      // the RDD from persistentRdds, and RDD.persist only re-registers on the way out of
+      // StorageLevel.NONE, so a reaped RDD stops appearing in getPersistentRDDs.
+      endpoint.rddReaper = { rddId =>
+        // Re-check the veto here, not just in rddReapable: an RDD can be locally checkpointed
+        // between the cleaner's check and this call, and reaping one destroys the only copy of its
+        // data. This narrows that window to the read below.
+        if (!isLocallyCheckpointedRdd(rddId)) {
+          _cleaner match {
+            case Some(contextCleaner) => contextCleaner.doCleanupRDD(rddId, blocking = false)
+            case None => unpersistRDD(rddId, blocking = false)
+          }
+        }
+      }
+    }
+
     if (_conf.get(UI_REVERSE_PROXY)) {
       val proxyUrl = _conf.get(UI_REVERSE_PROXY_URL).getOrElse("").stripSuffix("/")
       System.setProperty("spark.ui.proxyBase", proxyUrl + "/proxy/" + _applicationId)
@@ -2582,6 +2628,19 @@ class SparkContext(config: SparkConf) extends Logging {
       ShutdownHookManager.removeShutdownHook(_shutdownHookRef)
     }
 
+    // Before the ContextCleaner, the listener bus and the shuffle driver components below, all of
+    // which a reap in flight would use. Their own owners (SparkEnv, the endpoint's onStop) stop
+    // them too, but not until the very end of the teardown.
+    if (_env != null) {
+      Utils.tryLogNonFatalError {
+        _env.blockManagerMasterEndpoint.foreach(_.ttlCleaner.foreach(_.stop()))
+        _env.mapOutputTracker match {
+          case mapOutputTrackerMaster: MapOutputTrackerMaster =>
+            mapOutputTrackerMaster.ttlCleaner.foreach(_.stop())
+          case _ =>
+        }
+      }
+    }
     if (listenerBus != null) {
       Utils.tryLogNonFatalError {
         postApplicationEnd(exitCode)
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index d48640d9469b4..75c8d8c803939 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/core/src/main/scala/org/apache/spark/SparkEnv.scala
@@ -78,6 +78,11 @@ class SparkEnv (
     val outputCommitCoordinator: OutputCommitCoordinator,
     val conf: SparkConf) extends Logging {
 
+  // The driver's BlockManagerMasterEndpoint, kept so SparkContext can wire the RDD TTL cleaner's
+  // hooks onto it (they need pieces that do not exist yet when SparkEnv builds the endpoint).
+  // None on executors, which only hold a ref to it.
+  @volatile private[spark] var blockManagerMasterEndpoint: Option[BlockManagerMasterEndpoint] = None
+
   // The two shuffle managers are peers keyed by kind, not a default and an override: a shuffle is
   // routed to one or the other by its dependency type via `shuffleManagerFor`, so neither is ever
   // installed "behind" the other.
@@ -694,21 +699,27 @@ object SparkEnv extends Logging {
 
     // Mapping from block manager id to the block manager's information.
     val blockManagerInfo = new concurrent.TrieMap[BlockManagerId, BlockManagerInfo]()
+    // Only the driver builds the endpoint; an executor just looks up its ref.
+    val driverBlockManagerMasterEndpoint = if (isDriver) {
+      Some(new BlockManagerMasterEndpoint(
+        rpcEnv,
+        isLocal,
+        conf,
+        listenerBus,
+        // Already None unless the shuffle service is enabled, which is what the old conditional
+        // here re-checked.
+        externalShuffleClient,
+        blockManagerInfo,
+        mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
+        isDriver))
+    } else {
+      None
+    }
     val blockManagerMaster = new BlockManagerMaster(
       registerOrLookupEndpoint(
         BlockManagerMaster.DRIVER_ENDPOINT_NAME,
-        new BlockManagerMasterEndpoint(
-          rpcEnv,
-          isLocal,
-          conf,
-          listenerBus,
-          if (conf.get(config.SHUFFLE_SERVICE_ENABLED)) {
-            externalShuffleClient
-          } else {
-            None
-          }, blockManagerInfo,
-          mapOutputTracker.asInstanceOf[MapOutputTrackerMaster],
-          isDriver)),
+        // By-name, and only evaluated when isDriver, which is exactly when this is non-empty.
+        driverBlockManagerMasterEndpoint.get),
       registerOrLookupEndpoint(
         BlockManagerMaster.DRIVER_HEARTBEAT_ENDPOINT_NAME,
         new BlockManagerMasterHeartbeatEndpoint(rpcEnv, isLocal, blockManagerInfo)),
@@ -773,6 +784,8 @@ object SparkEnv extends Logging {
       outputCommitCoordinator,
       conf)
 
+    envInstance.blockManagerMasterEndpoint = driverBlockManagerMasterEndpoint
+
     // Add a reference to tmp dir created by driver, we will delete this tmp dir when stop() is
     // called, and we only need to do it for driver. Because driver may run as a service, and if we
     // don't delete this tmp dir when sc is stopped, then will create too many tmp dirs.
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 6303f751c9c14..13fc7110dd9b5 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
@@ -2207,6 +2207,37 @@ package object config {
       .version("4.2.0")
       .fallbackConf(Network.NETWORK_TIMEOUT)
 
+  // Both TTLs below are in seconds when given without a unit, matching the pre-2.0
+  // `spark.cleaner.ttl` they restore, and both refuse anything under 10 minutes: the TTL has to
+  // outlast the longest gap between uses and the longest stage runtime, and a value below that
+  // reclaims blocks a running job still needs.
+  private[spark] val CLEANER_TTL_MINIMUM_SECONDS = 600L
+
+  private[spark] val CLEANER_TTL_RDD =
+    ConfigBuilder("spark.cleaner.ttl.rdd")
+      .doc("Add a TTL for RDD cache blocks (broadcast blocks are not TTL-cleaned; shuffle blocks " +
+        "use spark.cleaner.ttl.shuffle). By default blocks are only removed after GC on the " +
+        "driver, which will not happen for DataFrames or RDDs at the global scope. Interpreted " +
+        "as seconds when no unit is given, and must be at least 10 minutes. Must be set before " +
+        "starting the SparkContext.")
+      .version("4.4.0")
+      .timeConf(TimeUnit.SECONDS)
+      .checkValue(_ >= CLEANER_TTL_MINIMUM_SECONDS,
+        "The RDD block TTL must be at least 10 minutes.")
+      .createOptional
+
+  private[spark] val CLEANER_TTL_SHUFFLE =
+    ConfigBuilder("spark.cleaner.ttl.shuffle")
+      .doc("Add a TTL for shuffle blocks. By default blocks are only removed after GC on the " +
+        "driver, which will not happen for DataFrames or RDDs at the global scope. Interpreted " +
+        "as seconds when no unit is given, and must be at least 10 minutes. Must be set before " +
+        "starting the SparkContext.")
+      .version("4.4.0")
+      .timeConf(TimeUnit.SECONDS)
+      .checkValue(_ >= CLEANER_TTL_MINIMUM_SECONDS,
+        "The shuffle block TTL must be at least 10 minutes.")
+      .createOptional
+
   private[spark] val EXECUTOR_LOGS_ROLLING_STRATEGY =
     ConfigBuilder("spark.executor.logs.rolling.strategy")
       .version("1.1.0")
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index e4cbbe6302af4..39fb739093ba9 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -1843,10 +1843,11 @@ abstract class RDD[T: ClassTag](
     checkpointData.exists(_.isCheckpointed)
 
   /**
-   * Return whether this RDD is marked for local checkpointing.
-   * Exposed for testing.
+   * Return whether this RDD is marked for local checkpointing. Also consulted by the RDD TTL
+   * cleaner, which must never reap a locally checkpointed RDD: local checkpointing truncates
+   * lineage, so its cache blocks are the only copy of the data.
    */
-  private[rdd] def isLocallyCheckpointed: Boolean = {
+  private[spark] def isLocallyCheckpointed: Boolean = {
     checkpointData match {
       case Some(_: LocalRDDCheckpointData[T]) => true
       case _ => false
@@ -1973,7 +1974,9 @@ abstract class RDD[T: ClassTag](
 
   private[spark] def elementClassTag: ClassTag[T] = classTag[T]
 
-  private[spark] var checkpointData: Option[RDDCheckpointData[T]] = None
+  // @volatile so the RDD TTL cleaner, which reads it via isLocallyCheckpointed from its own thread,
+  // sees a local checkpoint that was marked on the driver thread.
+  @volatile private[spark] var checkpointData: Option[RDDCheckpointData[T]] = None
 
   // Whether to checkpoint all ancestor RDDs that are marked for checkpointing. By default,
   // we stop as soon as we find the first such RDD, an optimization that allows us to write
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
index 47b6e2a50525f..b7f6449afcfe0 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMaster.scala
@@ -21,14 +21,15 @@ import scala.collection.BuildFrom
 import scala.collection.immutable.Iterable
 import scala.concurrent.Future
 
-import org.apache.spark.SparkConf
+import org.apache.spark.{MapOutputTrackerMaster, SparkConf, SparkEnv}
 import org.apache.spark.errors.SparkCoreErrors
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys._
+import org.apache.spark.internal.config
 import org.apache.spark.internal.config.CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT
 import org.apache.spark.rpc.{RpcEndpointRef, RpcTimeout}
 import org.apache.spark.storage.BlockManagerMessages._
-import org.apache.spark.util.{RpcUtils, ThreadUtils}
+import org.apache.spark.util.{RpcUtils, ThreadUtils, Utils}
 
 private[spark]
 class BlockManagerMaster(
@@ -43,6 +44,13 @@ class BlockManagerMaster(
   private val waitBlockRemovalTimeout =
     RpcTimeout(conf, CLEANER_REFERENCE_TRACKING_BLOCKING_TIMEOUT.key, "120s")
 
+  // Mirrors BlockManagerMasterEndpoint.externalShuffleServiceRemoveShuffleEnabled: the endpoint
+  // gates on its external block store client being defined, which is exactly when the shuffle
+  // service is enabled.
+  private val removeShuffleFromExternalServiceEnabled =
+    conf.get(config.SHUFFLE_SERVICE_ENABLED) &&
+      conf.get(config.SHUFFLE_SERVICE_REMOVE_SHUFFLE_ENABLED)
+
   /** Remove a dead executor from the driver endpoint. This is only called on the driver side. */
   def removeExecutor(execId: String): Unit = {
     tell(RemoveExecutor(execId))
@@ -234,7 +242,9 @@ class BlockManagerMaster(
 
   /** Remove all blocks belonging to the given shuffle. */
   def removeShuffle(shuffleId: Int, blocking: Boolean): Unit = {
-    val future = driverEndpoint.askSync[Future[Seq[Boolean]]](RemoveShuffle(shuffleId))
+    val (mapOutputLocations, shuffleMergerLocations) = shuffleOutputLocations(shuffleId)
+    val future = driverEndpoint.askSync[Future[Seq[Boolean]]](
+      RemoveShuffleFromMaster(shuffleId, mapOutputLocations, shuffleMergerLocations))
     future.failed.foreach(e =>
       logWarning(log"Failed to remove shuffle ${MDC(SHUFFLE_ID, shuffleId)} - " +
         log"${MDC(ERROR, e.getMessage)}", e)
@@ -244,6 +254,44 @@ class BlockManagerMaster(
     }
   }
 
+  /**
+   * Snapshots the shuffle output locations that `BlockManagerMasterEndpoint.removeShuffle` needs:
+   * the map output locations it matches against departed executors to find blocks the external
+   * shuffle service still serves, and the push-merger locations it asks the service to clean up.
+   *
+   * Read here, on the caller's thread, rather than on the master's dispatcher thread, because
+   * reading either takes the shuffle's `ShuffleStatus` lock. A thread holding that lock for write
+   * can be blocked on an RPC that only the block-manager dispatcher can serve -- serializing the
+   * map statuses broadcasts them, and the broadcast reports its blocks to the master; invalidating
+   * the cache destroys that broadcast, which asks the master to remove them. Acquiring the lock on
+   * the dispatcher closes the cycle, and since that dispatcher is single-threaded, it wedges all
+   * block management until the RPC times out. SPARK-36782 is the same cycle, fixed the same way for
+   * `updateMapOutput`. Snapshotting before the RPC also means the caller is free to unregister the
+   * shuffle as soon as `removeShuffle` returns.
+   *
+   * Both gates mirror the endpoint's own, so nothing is gathered that it would not use, and both
+   * are empty off the driver or before the `SparkEnv` exists.
+   */
+  private def shuffleOutputLocations(
+      shuffleId: Int): (Seq[(BlockManagerId, Long)], Seq[BlockManagerId]) = {
+    val tracker = Option(SparkEnv.get).map(_.mapOutputTracker).collect {
+      case master: MapOutputTrackerMaster => master
+    }
+    tracker.map { mapOutputTracker =>
+      val mapOutputLocations = if (removeShuffleFromExternalServiceEnabled) {
+        mapOutputTracker.getMapOutputLocations(shuffleId)
+      } else {
+        Seq.empty
+      }
+      val shuffleMergerLocations = if (Utils.isPushBasedShuffleEnabled(conf, isDriver)) {
+        mapOutputTracker.getShufflePushMergerLocations(shuffleId)
+      } else {
+        Seq.empty
+      }
+      (mapOutputLocations, shuffleMergerLocations)
+    }.getOrElse((Seq.empty, Seq.empty))
+  }
+
   /** Remove all blocks belonging to the given broadcast. */
   def removeBroadcast(broadcastId: Long, removeFromMaster: Boolean, blocking: Boolean): Unit = {
     val future = driverEndpoint.askSync[Future[Seq[Int]]](
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
index beca0d11d1c5e..45fafff389420 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMasterEndpoint.scala
@@ -19,7 +19,7 @@ package org.apache.spark.storage
 
 import java.io.IOException
 import java.util.{HashMap => JHashMap}
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
 
 import scala.collection.mutable
 import scala.concurrent.{ExecutionContext, ExecutionContextExecutorService, Future, TimeoutException}
@@ -29,7 +29,7 @@ import scala.util.control.NonFatal
 
 import com.google.common.cache.CacheBuilder
 
-import org.apache.spark.{MapOutputTrackerMaster, SparkConf, SparkContext, SparkEnv}
+import org.apache.spark.{BlockTTLCleaner, MapOutputTrackerMaster, SparkConf, SparkContext, SparkEnv}
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.internal.{config, Logging}
 import org.apache.spark.internal.LogKeys._
@@ -115,6 +115,39 @@ class BlockManagerMasterEndpoint(
   private implicit val askExecutionContext: ExecutionContextExecutorService =
     ExecutionContext.fromExecutorService(askThreadPool)
 
+  // Last access time per RDD, for the RDD TTL. Overwriting a "close" timestamp is fine, but the TTL
+  // cleaner thread iterates and removes from this while the dispatcher puts to it, so it has to be
+  // a concurrent map.
+  private[spark] val rddAccessTime = new ConcurrentHashMap[Int, Long]
+
+  // Read once: updateBlockAtime is on the hot path of block location lookups. The config is in
+  // seconds; the cleaner works in millis, like the access times it compares against.
+  private val rddTtl: Option[Long] =
+    conf.get(config.CLEANER_TTL_RDD).map(TimeUnit.SECONDS.toMillis)
+
+  // Veto consulted before the TTL cleaner reaps an RDD. Wired by SparkContext to refuse
+  // locally-checkpointed RDDs, whose cache blocks are the only copy of their data. Defaults to
+  // refusing everything: reaping one of those loses data, so an unwired cleaner must reap nothing.
+  @volatile private[spark] var rddReapable: Int => Boolean = _ => false
+
+  // How the TTL cleaner reaps an RDD; wired by SparkContext to ContextCleaner.doCleanupRDD, which
+  // removes the blocks and runs the CleanerListener fan-out. Never called while unwired, since
+  // rddReapable refuses everything until SparkContext wires both.
+  @volatile private[spark] var rddReaper: Int => Unit = _ => ()
+
+  // Started in onStart rather than here, once the endpoint is live and SparkContext has had its
+  // chance to wire the two hooks above. Both are read through a lambda for that reason: the vars
+  // are assigned after this cleaner is built.
+  private[spark] val ttlCleaner: Option[BlockTTLCleaner] = rddTtl.map { ttl =>
+    new BlockTTLCleaner(
+      name = "RDD",
+      idKey = RDD_ID,
+      ttlMillis = ttl,
+      accessTimes = rddAccessTime,
+      shouldReap = rddId => rddReapable(rddId),
+      reap = rddId => rddReaper(rddId))
+  }
+
   private val topologyMapper = {
     val topologyMapperClassName = conf.get(
       config.STORAGE_REPLICATION_TOPOLOGY_MAPPER)
@@ -209,8 +242,8 @@ class BlockManagerMasterEndpoint(
     case RemoveRdd(rddId) =>
       context.reply(removeRdd(rddId))
 
-    case RemoveShuffle(shuffleId) =>
-      context.reply(removeShuffle(shuffleId))
+    case RemoveShuffleFromMaster(shuffleId, mapOutputLocations, shuffleMergerLocations) =>
+      context.reply(removeShuffle(shuffleId, mapOutputLocations, shuffleMergerLocations))
 
     case RemoveBroadcast(broadcastId, removeFromDriver) =>
       context.reply(removeBroadcast(broadcastId, removeFromDriver))
@@ -270,6 +303,22 @@ class BlockManagerMasterEndpoint(
       context.reply(updateRDDBlockVisibility(taskId, visible))
   }
 
+  // Only RDD cache blocks are TTL-tracked here; shuffle atimes are the map output tracker's job,
+  // and broadcast blocks are not TTL-cleaned at all.
+  private def updateBlockAtime(blockId: BlockId): Unit = {
+    // This is on the dispatcher thread for every block location lookup (O(partitions) per stage via
+    // DAGScheduler.getCacheLocs), so do no work when the TTL is unset, which is the default.
+    if (rddTtl.isDefined) {
+      blockId match {
+        // Only stamp a block we actually track: a lookup of an absent block would otherwise leave a
+        // phantom atime, and one TTL later a pointless cluster-wide RemoveRdd.
+        case rddBlockId: RDDBlockId if blockLocations.containsKey(blockId) =>
+          rddAccessTime.put(rddBlockId.rddId, System.currentTimeMillis())
+        case _ =>
+      }
+    }
+  }
+
   private def isRDDBlockVisible(blockId: RDDBlockId): Boolean = {
     if (trackingCacheVisibility) {
       blockLocations.containsKey(blockId) &&
@@ -366,6 +415,8 @@ class BlockManagerMasterEndpoint(
   }
 
   private def removeRdd(rddId: Int): Future[Seq[Int]] = {
+    rddAccessTime.remove(rddId)
+
     // First remove the metadata for the given RDD, and then asynchronously remove the blocks
     // from the storage endpoints.
 
@@ -437,29 +488,32 @@ class BlockManagerMasterEndpoint(
     Future.sequence(removeRddFromExecutorsFutures ++ removeRddBlockViaExtShuffleServiceFutures)
   }
 
-  private def removeShuffle(shuffleId: Int): Future[Seq[Boolean]] = {
-    // Find all shuffle blocks on executors that are no longer running
+  // `mapOutputLocations` and `shuffleMergerLocations` are snapshotted by the caller rather than
+  // read from the MapOutputTracker here, because reading either takes the shuffle's ShuffleStatus
+  // lock and this is the single dispatcher thread. See BlockManagerMaster.removeShuffle.
+  private def removeShuffle(
+      shuffleId: Int,
+      mapOutputLocations: Seq[(BlockManagerId, Long)],
+      shuffleMergerLocations: Seq[BlockManagerId]): Future[Seq[Boolean]] = {
+    // Find all shuffle blocks on executors that are no longer running. No gate on
+    // externalShuffleServiceRemoveShuffleEnabled here: the caller applies that gate when deciding
+    // whether to gather the snapshot, so re-deriving it would just be a second copy of the
+    // condition that could disagree with the first and silently skip the removal.
     val blocksToDeleteByShuffleService =
       new mutable.HashMap[BlockManagerId, mutable.HashSet[BlockId]]
-    if (externalShuffleServiceRemoveShuffleEnabled) {
-      mapOutputTracker.shuffleStatuses.get(shuffleId).foreach { shuffleStatus =>
-        shuffleStatus.withMapStatuses { mapStatuses =>
-          mapStatuses.filter(_ != null).foreach { mapStatus =>
-            // Check if the executor has been deallocated
-            if (!blockManagerIdByExecutor.contains(mapStatus.location.executorId)) {
-              // Only a BlockingShuffleManager serves block-manager blocks; a pipelined shuffle is
-              // served out-of-band and is not in the MapOutputTracker this loop iterates, so this
-              // resolves only regular shuffles (None only before the manager is initialized).
-              val blocksToDel = SparkEnv.get.shuffleBlockResolver
-                .map(_.getBlocksForShuffle(shuffleId, mapStatus.mapId))
-                .getOrElse(Seq.empty)
-              if (blocksToDel.nonEmpty) {
-                val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(mapStatus.location,
-                  new mutable.HashSet[BlockId])
-                blocks ++= blocksToDel
-              }
-            }
-          }
+    mapOutputLocations.foreach { case (location, mapId) =>
+      // Check if the executor has been deallocated
+      if (!blockManagerIdByExecutor.contains(location.executorId)) {
+        // Only a BlockingShuffleManager serves block-manager blocks; a pipelined shuffle is
+        // served out-of-band and is not in the MapOutputTracker the snapshot came from, so this
+        // resolves only regular shuffles (None only before the manager is initialized).
+        val blocksToDel = SparkEnv.get.shuffleBlockResolver
+          .map(_.getBlocksForShuffle(shuffleId, mapId))
+          .getOrElse(Seq.empty)
+        if (blocksToDel.nonEmpty) {
+          val blocks = blocksToDeleteByShuffleService.getOrElseUpdate(location,
+            new mutable.HashSet[BlockId])
+          blocks ++= blocksToDel
         }
       }
     }
@@ -481,13 +535,8 @@ class BlockManagerMasterEndpoint(
 
     val removeShuffleMergeFromShuffleServicesFutures =
       externalBlockStoreClient.map { shuffleClient =>
-        val mergerLocations =
-          if (Utils.isPushBasedShuffleEnabled(conf, isDriver)) {
-            mapOutputTracker.getShufflePushMergerLocations(shuffleId)
-          } else {
-            Seq.empty[BlockManagerId]
-          }
-        mergerLocations.map { bmId =>
+        // Likewise gated by the caller, which only gathers these under push-based shuffle.
+        shuffleMergerLocations.map { bmId =>
           Future[Boolean] {
             shuffleClient.removeShuffleMerge(bmId.host, bmId.port, shuffleId,
               RemoteBlockPushResolver.DELETE_ALL_MERGED_SHUFFLE)
@@ -857,6 +906,12 @@ class BlockManagerMasterEndpoint(
     } else {
       locations = new mutable.HashSet[BlockManagerId]
       blockLocations.put(blockId, locations)
+      // The initial put counts as an access, but only when the report actually stores the block: an
+      // invalid level here is a *removal* report (e.g. the replies to a RemoveRdd broadcast), and
+      // stamping those would resurrect the atime of an RDD with no blocks left.
+      if (storageLevel.isValid) {
+        updateBlockAtime(blockId)
+      }
     }
 
     if (storageLevel.isValid) {
@@ -1012,12 +1067,14 @@ class BlockManagerMasterEndpoint(
   }
 
   private def getLocations(blockId: BlockId): Seq[BlockManagerId] = {
+    updateBlockAtime(blockId)
     if (blockLocations.containsKey(blockId)) blockLocations.get(blockId).toSeq else Seq.empty
   }
 
   private def getLocationsAndStatus(
       blockId: BlockId,
       requesterHost: String): Option[BlockLocationsAndStatus] = {
+    updateBlockAtime(blockId)
     val allLocations = Option(blockLocations.get(blockId)).map(_.toSeq).getOrElse(Seq.empty)
     val blockStatusWithBlockManagerId: Option[(BlockStatus, BlockManagerId)] =
       (if (externalShuffleServiceRddFetchEnabled && blockId.isRDD) {
@@ -1130,7 +1187,16 @@ class BlockManagerMasterEndpoint(
     }
   }
 
+  // The reap itself stays off this thread but its removal does not: rddReaper routes through
+  // BlockManagerMaster.removeRdd, i.e. a RemoveRdd message back to this endpoint, so blockLocations
+  // and friends are still only touched by the dispatcher (the IsolatedThreadSafeRpcEndpoint
+  // invariant).
+  override def onStart(): Unit = ttlCleaner.foreach(_.start())
+
   override def onStop(): Unit = {
+    // Cleaner first: a reap in flight routes its removal through askThreadPool, so shutting that
+    // down before joining the cleaner would fail the reap with a RejectedExecutionException.
+    ttlCleaner.foreach(_.stop())
     askThreadPool.shutdownNow()
   }
 }
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
index 3c2ee19ffbd74..6357786afac28 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerMessages.scala
@@ -44,6 +44,16 @@ private[spark] object BlockManagerMessages {
   // Remove all blocks belonging to a specific shuffle.
   case class RemoveShuffle(shuffleId: Int) extends ToBlockManagerMasterStorageEndpoint
 
+  // Driver-side entry point for removing a shuffle, sent only by BlockManagerMaster.removeShuffle.
+  // Distinct from RemoveShuffle, which is what the master then fans out to the storage endpoints,
+  // because these two payloads are of no use to an executor. They are snapshotted by the caller
+  // rather than read on the master's dispatcher thread; see BlockManagerMaster.removeShuffle for
+  // why that matters. Both are empty unless the corresponding feature is on.
+  case class RemoveShuffleFromMaster(
+      shuffleId: Int,
+      mapOutputLocations: Seq[(BlockManagerId, Long)],
+      shuffleMergerLocations: Seq[BlockManagerId]) extends ToBlockManagerMaster
+
   // Remove all blocks belonging to a specific broadcast.
   case class RemoveBroadcast(broadcastId: Long, removeFromDriver: Boolean = true)
     extends ToBlockManagerMasterStorageEndpoint
diff --git a/core/src/test/scala/org/apache/spark/BlockTTLCleanerSuite.scala b/core/src/test/scala/org/apache/spark/BlockTTLCleanerSuite.scala
new file mode 100644
index 0000000000000..382317cc08a59
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/BlockTTLCleanerSuite.scala
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark
+
+import java.util.concurrent.{ConcurrentHashMap, TimeUnit}
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.scalatest.concurrent.Eventually
+import org.scalatest.time.{Seconds, Span}
+
+import org.apache.spark.internal.LogKeys.RDD_ID
+
+/**
+ * Unit tests for the TTL sweep itself. `BlockTTLIntegrationSuite` covers the wiring end to end
+ * against a real cluster; this drives `BlockTTLCleaner` directly with a plain map, veto and reap,
+ * so the parts that are awkward to provoke through a SparkContext -- a vetoed id, an access time
+ * that moves mid-sweep, a reap that throws -- are cheap to pin down.
+ *
+ * Most tests call `sweep()` rather than starting the thread, so there is nothing to wait for.
+ */
+class BlockTTLCleanerSuite extends SparkFunSuite with Eventually {
+
+  private val ttlMillis = 10000L
+
+  private def cleaner(
+      accessTimes: ConcurrentHashMap[Int, Long],
+      shouldReap: Int => Boolean = _ => true)(
+      reap: Int => Unit): BlockTTLCleaner =
+    new BlockTTLCleaner("test", RDD_ID, ttlMillis, accessTimes, shouldReap, reap)
+
+  /** An access time far enough in the past that a sweep sees it as expired. */
+  private def expired: Long = System.currentTimeMillis() - (ttlMillis * 10)
+
+  test("an idle id is reaped and left untracked") {
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    accessTimes.put(7, expired)
+    val reaped = ArrayBuffer.empty[Int]
+    cleaner(accessTimes)(reaped += _).sweep()
+    assert(reaped.toSeq === Seq(7))
+    assert(accessTimes.isEmpty, "a reaped id should not stay tracked")
+  }
+
+  test("a live id is left alone") {
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    accessTimes.put(3, System.currentTimeMillis())
+    accessTimes.put(5, expired)
+    val reaped = ArrayBuffer.empty[Int]
+    cleaner(accessTimes)(reaped += _).sweep()
+    assert(reaped.toSeq === Seq(5))
+    assert(accessTimes.containsKey(3), "an id inside its TTL must not be reaped")
+  }
+
+  test("a vetoed id is never reaped but stays tracked for a later sweep") {
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    accessTimes.put(1, expired)
+    accessTimes.put(2, expired)
+    val reaped = ArrayBuffer.empty[Int]
+    // 1 is vetoed (a locally-checkpointed RDD, or a shuffle with no output yet), 2 is not.
+    val c = cleaner(accessTimes, shouldReap = _ != 1)(reaped += _)
+    c.sweep()
+    c.sweep()
+    assert(reaped.toSeq === Seq(2), "a vetoed id must not be reaped, on this sweep or a later one")
+    assert(accessTimes.containsKey(1),
+      "a vetoed id must stay tracked, or it could never be reaped once the veto lifts")
+  }
+
+  test("an id that comes back into use between the scan and the reap is not reaped") {
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    accessTimes.put(9, expired)
+    val reaped = new AtomicInteger(0)
+    // shouldReap is the sweep's last look before it commits, so refreshing the access time here
+    // reproduces a use landing in exactly that window. The compare-and-remove must then fail.
+    cleaner(accessTimes, shouldReap = { _ =>
+      accessTimes.put(9, System.currentTimeMillis())
+      true
+    })(_ => reaped.incrementAndGet()).sweep()
+    assert(reaped.get() === 0,
+      "the access time moved after the scan read it, so the compare-and-remove should have failed")
+    assert(accessTimes.containsKey(9))
+  }
+
+  test("a failed reap leaves the id tracked so a later sweep retries it") {
+    // Regression test: dropping the access time before the reap meant a reap that threw part-way (a
+    // blocking shuffle removal timing out, say) left the id untracked forever, so whatever it still
+    // held on disk leaked for the life of the driver.
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    accessTimes.put(4, expired)
+    val attempts = new AtomicInteger(0)
+    val c = cleaner(accessTimes) { _ =>
+      attempts.incrementAndGet()
+      throw new IllegalStateException("boom")
+    }
+    c.sweep()
+    assert(attempts.get() === 1)
+    assert(accessTimes.containsKey(4), "a failed reap must leave the id tracked")
+    // The retry is one TTL out, not the very next sweep, so a permanently failing reap costs one
+    // warning per TTL rather than one per sweep.
+    c.sweep()
+    assert(attempts.get() === 1, "the restored access time should be fresh, not still expired")
+    accessTimes.put(4, expired)
+    c.sweep()
+    assert(attempts.get() === 2, "once expired again, the id should be retried")
+  }
+
+  test("the sweep asks to be woken before the oldest live entry expires") {
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    // Half a TTL old, so it expires in roughly half a TTL from now.
+    accessTimes.put(1, System.currentTimeMillis() - (ttlMillis / 2))
+    val wait = cleaner(accessTimes)(_ => ()).sweep()
+    assert(wait <= ttlMillis / 2 + 1000, s"should wake near the entry's expiry, not in $wait ms")
+    assert(wait >= ttlMillis / 10, "but never sooner than the floor, or the sweep would spin")
+  }
+
+  test("an empty map waits a whole TTL rather than spinning") {
+    val wait = cleaner(new ConcurrentHashMap[Int, Long])(_ => ()).sweep()
+    assert(wait >= ttlMillis - 1000 && wait <= ttlMillis)
+  }
+
+  test("start and stop run and end the sweep thread") {
+    val accessTimes = new ConcurrentHashMap[Int, Long]
+    accessTimes.put(1, expired)
+    val reaped = new AtomicInteger(0)
+    val c = cleaner(accessTimes)(_ => reaped.incrementAndGet())
+    c.start()
+    // The first sweep happens immediately, so this needs no TTL-length wait.
+    eventually(timeout(Span(10, Seconds))) {
+      assert(reaped.get() === 1)
+    }
+    c.stop()
+    accessTimes.put(2, expired)
+    // Well over the floor (ttl/10 = 1s), so a still-running sweep would have taken this.
+    TimeUnit.SECONDS.sleep(3)
+    assert(accessTimes.containsKey(2), "a stopped cleaner must not reap anything")
+  }
+}
diff --git a/core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala b/core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala
new file mode 100644
index 0000000000000..12a529276c32d
--- /dev/null
+++ b/core/src/test/scala/org/apache/spark/storage/BlockTTLIntegrationSuite.scala
@@ -0,0 +1,271 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.storage
+
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.scalatest.concurrent.Eventually
+import org.scalatest.time._
+
+import org.apache.spark._
+import org.apache.spark.internal.config
+
+/**
+ * End-to-end tests for the block TTL cleaners: that the right things get tracked, that a reap
+ * reclaims what it should, and that the hooks `SparkContext` wires are the ones that run.
+ *
+ * The smallest TTL the configs accept is 10 minutes, so these tests do not wait for an expiry.
+ * Instead they backdate an access time and call `BlockTTLCleaner.sweep()` directly, which is both
+ * faster and deterministic. The background sweep threads are stopped for the same reason -- their
+ * scheduling is `BlockTTLCleanerSuite`'s business, not this suite's.
+ */
+class BlockTTLIntegrationSuite extends SparkFunSuite with LocalSparkContext with Eventually {
+
+  // Only the cluster-wide block removals are asynchronous now, so the budget can be modest. The
+  // interval is deliberately not tight: some `eventually` bodies do an RPC fan-out to every
+  // executor, and polling that every few millis loads the dispatcher being measured.
+  implicit override val patienceConfig: PatienceConfig =
+    PatienceConfig(timeout = scaled(Span(20, Seconds)), interval = scaled(Span(200, Millis)))
+
+  private val ttlSeconds = 600L
+  private val numParts = 3
+
+  /** Starts a two-executor cluster, enabling either cleaner. */
+  private def startCluster(rddTTL: Boolean, shuffleTTL: Boolean): Unit = {
+    val conf = new SparkConf()
+      .setAppName(getClass.getSimpleName)
+      .setMaster("local-cluster[2, 1, 1024]")
+    if (rddTTL) conf.set(config.CLEANER_TTL_RDD, ttlSeconds)
+    if (shuffleTTL) conf.set(config.CLEANER_TTL_SHUFFLE, ttlSeconds)
+    sc = new SparkContext(conf)
+    TestUtils.waitUntilExecutorsUp(sc, 2, 60000)
+    // Sweeps are driven explicitly below; stop the threads so they cannot race the assertions.
+    endpoint.ttlCleaner.foreach(_.stop())
+    tracker.ttlCleaner.foreach(_.stop())
+  }
+
+  // The driver retains its endpoint instance, so no reflection is needed to reach it.
+  private def endpoint: BlockManagerMasterEndpoint = sc.env.blockManagerMasterEndpoint.get
+
+  private def tracker: MapOutputTrackerMaster =
+    sc.env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
+
+  /** Backdates an access time well past the TTL, so the next sweep treats the id as expired. */
+  private def expire(accessTimes: ConcurrentHashMap[Int, Long], id: Int): Unit = {
+    assert(accessTimes.containsKey(id), s"$id should be TTL-tracked before being expired")
+    accessTimes.put(id, System.currentTimeMillis() - (ttlSeconds * 1000 * 2))
+  }
+
+  private def sweepRdds(): Unit =
+    endpoint.ttlCleaner.getOrElse(fail("the RDD TTL cleaner should be enabled")).sweep()
+
+  private def sweepShuffles(): Unit =
+    tracker.ttlCleaner.getOrElse(fail("the shuffle TTL cleaner should be enabled")).sweep()
+
+  test("a shuffle's map output is reclaimed by a sweep") {
+    startCluster(rddTTL = true, shuffleTTL = true)
+    assert(endpoint.rddAccessTime.isEmpty)
+    assert(tracker.shuffleAccessTime.isEmpty)
+    val shuffled = sc.parallelize(1.to(100)).groupBy(_ % 10)
+    shuffled.count()
+    val shuffleId = tracker.shuffleStatuses.keys.head
+    val epochBefore = tracker.getEpoch
+    eventually { assert(tracker.shuffleAccessTime.containsKey(shuffleId)) }
+    // An uncached shuffle is tracked by the map output tracker only, never as an RDD.
+    assert(endpoint.rddAccessTime.isEmpty)
+    assert(tracker.shuffleStatuses.get(shuffleId).exists(_.numAvailableMapOutputs > 0))
+
+    expire(tracker.shuffleAccessTime, shuffleId)
+    sweepShuffles()
+
+    // Assert on the map output itself, not just on the tracking map: the sweep drops the access
+    // time *before* it reaps, so an access-time assertion alone passes even with the reap wiring
+    // removed.
+    assert(!tracker.shuffleStatuses.get(shuffleId).exists(_.numAvailableMapOutputs > 0),
+      "the reaped shuffle's map output should be gone")
+    assert(tracker.getEpoch > epochBefore,
+      "the reap must bump the epoch, or executors keep fetching the files it deleted")
+    assert(tracker.shuffleAccessTime.isEmpty)
+    // Last use of `shuffled`, keeping it reachable throughout: the JVM collects by liveness, not by
+    // scope, and a collected ShuffleDependency would let the ContextCleaner do the reaping instead.
+    assert(shuffled.getNumPartitions > 0)
+  }
+
+  test("the shuffle reap deletes files before it drops the driver's map output") {
+    // removeShuffle reads the live MapStatuses and push-merger locations to find what the external
+    // shuffle service must delete, so emptying the ShuffleStatus first silently reclaims no ESS
+    // disk. Observe the ordering directly rather than standing up a shuffle service.
+    startCluster(rddTTL = false, shuffleTTL = true)
+    // -2 = never invoked, -1 = invoked with the ShuffleStatus already gone.
+    val outputsWhenRemoved = new AtomicInteger(-2)
+    val wiredRemover = tracker.shuffleFileRemover.getOrElse(
+      fail("SparkContext should have wired the shuffle file remover"))
+    tracker.shuffleFileRemover = Some { shuffleId =>
+      outputsWhenRemoved.compareAndSet(-2,
+        tracker.shuffleStatuses.get(shuffleId).map(_.numAvailableMapOutputs).getOrElse(-1))
+      wiredRemover(shuffleId)
+    }
+    val shuffled = sc.parallelize(1.to(100)).groupBy(_ % 10)
+    shuffled.count()
+    val shuffleId = tracker.shuffleStatuses.keys.head
+    eventually { assert(tracker.shuffleAccessTime.containsKey(shuffleId)) }
+
+    expire(tracker.shuffleAccessTime, shuffleId)
+    sweepShuffles()
+
+    assert(outputsWhenRemoved.get() !== -2, "the sweep should have reaped the shuffle")
+    assert(outputsWhenRemoved.get() > 0,
+      "the shuffle's map output must still be registered when its files are deleted, or the " +
+        "external shuffle service is never told which blocks to remove")
+    assert(shuffled.getNumPartitions > 0)
+  }
+
+  test("re-reading a cached RDD in a new job refreshes its access time") {
+    // This pins down the assumption behind the RDD TTL: an actively reused cached RDD is not
+    // reaped, because every job that reads it re-resolves cache locations at the driver
+    // (DAGScheduler.getCacheLocs -> BlockManagerMaster.getLocations -> updateBlockAtime), whether
+    // or not the block read itself is served locally on the executor.
+    startCluster(rddTTL = true, shuffleTTL = false)
+    assert(endpoint.rddAccessTime.isEmpty)
+    val input = sc.parallelize(1.to(100), numParts).cache()
+    input.count()
+    // eventually because the executors' UpdateBlockInfo reports can land after count() returns.
+    eventually { assert(endpoint.rddAccessTime.containsKey(input.id)) }
+    // Read via Option: a ConcurrentHashMap[Int, Long] miss unboxes null to 0L rather than throwing,
+    // which would let the assertion below pass vacuously.
+    def atimeOf(rddId: Int): Option[Long] =
+      Option(endpoint.rddAccessTime.get(rddId)).map(_.longValue)
+    val firstAtime = atimeOf(input.id).getOrElse(fail("the cached RDD should be TTL-tracked"))
+    // Retried because the clock may not have ticked past firstAtime yet.
+    eventually {
+      input.count()
+      assert(atimeOf(input.id).exists(_ > firstAtime),
+        s"a new job reading the cached RDD should refresh its atime (was $firstAtime)")
+    }
+  }
+
+  test("RDD and shuffle blocks are both reclaimed, and come back") {
+    startCluster(rddTTL = true, shuffleTTL = true)
+    assert(endpoint.rddAccessTime.isEmpty)
+    val input = sc.parallelize(1.to(100)).groupBy(_ % 10).cache()
+    input.count()
+    eventually { assert(endpoint.rddAccessTime.containsKey(input.id)) }
+    val shuffleId = tracker.shuffleStatuses.keys.head
+    eventually { assert(tracker.shuffleAccessTime.containsKey(shuffleId)) }
+
+    expire(endpoint.rddAccessTime, input.id)
+    expire(tracker.shuffleAccessTime, shuffleId)
+    sweepRdds()
+    sweepShuffles()
+
+    assert(tracker.shuffleAccessTime.isEmpty, "the reaped shuffle should no longer be tracked")
+    assert(endpoint.rddAccessTime.isEmpty, "the reaped RDD should no longer be tracked")
+    // Both are evictions, not unpersists: the job runs again and both come back tracked.
+    input.count()
+    eventually {
+      assert(endpoint.rddAccessTime.containsKey(input.id))
+      assert(!tracker.shuffleAccessTime.isEmpty)
+    }
+  }
+
+  test("a locally-checkpointed RDD is never reaped by the TTL cleaner") {
+    // localCheckpoint truncates lineage, so the cache blocks are the only copy of the data and
+    // losing them is unrecoverable (LocalCheckpointRDD.compute always throws).
+    startCluster(rddTTL = true, shuffleTTL = false)
+    val checkpointed = sc.parallelize(1.to(100), numParts)
+    checkpointed.localCheckpoint()
+    assert(checkpointed.count() === 100)
+    eventually { assert(endpoint.rddAccessTime.containsKey(checkpointed.id)) }
+    assert(endpoint.rddReapable(checkpointed.id) === false,
+      "the TTL cleaner must refuse to reap a locally-checkpointed RDD")
+
+    // Expire it and sweep anyway: the veto, not the access time, is what has to save it.
+    expire(endpoint.rddAccessTime, checkpointed.id)
+    sweepRdds()
+
+    assert(checkpointed.count() === 100,
+      "a locally-checkpointed RDD must survive a sweep: its blocks are the only copy")
+    assert(endpoint.rddAccessTime.containsKey(checkpointed.id),
+      "a vetoed RDD must stay tracked, or it could never be reaped once the veto lifts")
+
+    // Positive control: `rddReapable`'s unwired default is `_ => false`, so the assertion above
+    // passes just as well when SparkContext never wired the veto at all. Pin down that the wired
+    // veto discriminates rather than refusing everything.
+    val plainCached = sc.parallelize(1.to(10)).cache()
+    assert(plainCached.count() === 10)
+    assert(endpoint.rddReapable(plainCached.id),
+      "the veto must be specific to locally-checkpointed RDDs, not a blanket refusal")
+  }
+
+  test("reaping an RDD removes every partition and leaves the RDD usable") {
+    // Access times are recorded per RDD id but stamped by individual block accesses, and a reap
+    // removes the whole RDD. Pin down both halves of that.
+    startCluster(rddTTL = true, shuffleTTL = false)
+    val input = sc.parallelize(1.to(100), numParts).cache()
+    assert(input.count() === 100)
+
+    def cachedPartitionsOf(rddId: Int): Int =
+      sc.env.blockManager.master.getMatchingBlockIds({
+        case RDDBlockId(id, _) => id == rddId
+        case _ => false
+      }, askStorageEndpoints = true).size
+
+    eventually {
+      assert(endpoint.rddAccessTime.containsKey(input.id))
+      assert(cachedPartitionsOf(input.id) >= numParts, s"expected all $numParts partitions cached")
+    }
+
+    expire(endpoint.rddAccessTime, input.id)
+    sweepRdds()
+
+    assert(!endpoint.rddAccessTime.containsKey(input.id),
+      "the reaped RDD should no longer be TTL-tracked")
+    // eventually: the reap frees the driver's metadata synchronously but the executors are told
+    // asynchronously (removeRdd with blocking = false).
+    eventually {
+      assert(cachedPartitionsOf(input.id) === 0,
+        "every partition of the reaped RDD should be removed, not just the idle ones")
+    }
+    // An eviction, not an unpersist: the RDD still computes and comes back tracked.
+    assert(input.count() === 100, "a reaped RDD must still be usable (recomputed)")
+    eventually { assert(endpoint.rddAccessTime.containsKey(input.id)) }
+  }
+
+  test("nothing is tracked when the TTLs are not set") {
+    startCluster(rddTTL = false, shuffleTTL = false)
+    sc.parallelize(1.to(100)).groupBy(_ % 10).cache().count()
+    assert(endpoint.rddAccessTime.isEmpty)
+    assert(tracker.shuffleAccessTime.isEmpty)
+    assert(endpoint.ttlCleaner.isEmpty, "no RDD TTL means no cleaner at all")
+    assert(tracker.ttlCleaner.isEmpty, "no shuffle TTL means no cleaner at all")
+  }
+
+  test("a TTL below the ten minute minimum is rejected") {
+    val conf = new SparkConf().set(config.CLEANER_TTL_RDD.key, "60s")
+    val e = intercept[IllegalArgumentException](conf.get(config.CLEANER_TTL_RDD))
+    assert(e.getMessage.contains("at least 10 minutes"))
+    val shuffleConf = new SparkConf().set(config.CLEANER_TTL_SHUFFLE.key, "5s")
+    val se = intercept[IllegalArgumentException](shuffleConf.get(config.CLEANER_TTL_SHUFFLE))
+    assert(se.getMessage.contains("at least 10 minutes"))
+    // A bare number is seconds, matching the pre-2.0 spark.cleaner.ttl this restores, so 600 is
+    // exactly the minimum and must be accepted.
+    assert(new SparkConf().set(config.CLEANER_TTL_RDD.key, "600")
+      .get(config.CLEANER_TTL_RDD).contains(600L))
+  }
+}
diff --git a/docs/configuration.md b/docs/configuration.md
index 616367592b626..67d99dbfab059 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -2281,6 +2281,41 @@ Apart from these, the following properties are also available, and may be useful
   </td>
   <td>1.4.0</td>
 </tr>
+<tr>
+  <td><code>spark.cleaner.ttl.rdd</code></td>
+  <td>(none)</td>
+  <td>
+    If set, RDD cache blocks that have not been accessed for this duration are removed. Interpreted
+    as seconds when no unit is given, and must be at least 10 minutes. By default cached blocks are
+    only removed once the RDD is garbage collected on the driver, which does not happen for a
+    DataFrame or RDD held at global scope. Broadcast blocks are not covered; shuffle blocks use
+    <code>spark.cleaner.ttl.shuffle</code>. An access is only recorded when the driver resolves the
+    block's locations, so set this comfortably longer than both the longest gap between uses and the
+    longest stage runtime, or the RDD will be removed and recomputed. Removal frees the blocks but
+    does not reset the RDD's storage level, so a later action re-caches it; the RDD does stop
+    appearing in <code>SparkContext.getPersistentRDDs</code>. Locally checkpointed RDDs are never
+    removed, since local checkpointing truncates lineage and leaves their cache blocks as the only
+    copy of the data. Must be set before the SparkContext is created.
+  </td>
+  <td>4.4.0</td>
+</tr>
+<tr>
+  <td><code>spark.cleaner.ttl.shuffle</code></td>
+  <td>(none)</td>
+  <td>
+    If set, shuffles that have not been accessed for this duration have their map output removed from
+    the driver and their shuffle files deleted on the executors. Interpreted as seconds when no unit
+    is given, and must be at least 10 minutes. By default shuffle data is only removed once the
+    shuffle dependency is garbage collected on the driver. An access is only recorded when a map task
+    registers output or the driver serves map output locations, and executors cache those locations,
+    so it is on you to set this longer than any shuffle's useful life: longer than both the longest
+    gap between uses and the longest stage runtime. Removing a shuffle that a running job still needs
+    fails that job -- its readers get a fetch failure, and the map stage cannot re-register its
+    output, so the stage is aborted rather than recomputed -- and no amount of retrying recovers it.
+    Must be set before the SparkContext is created.
+  </td>
+  <td>4.4.0</td>
+</tr>
 </table>
 
 ### Execution Behavior