Restore incrementEpoch() call.

Author: JoshRosen

core/src/main/scala/org/apache/spark/MapOutputTracker.scala

@@ -532,7 +532,7 @@ private[spark] class MapOutputTrackerMaster(
     None
   }
 
-  private def incrementEpoch() {
+  def incrementEpoch() {
     epochLock.synchronized {
       epoch += 1
       logDebug("Increasing epoch to " + epoch)

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -1224,6 +1224,15 @@ class DAGScheduler(
               logInfo("waiting: " + waitingStages)
               logInfo("failed: " + failedStages)
 
+              // This call to increment the epoch may not be strictly necessary, but it is retained
+              // for now in order to minimize the changes in behavior from an earlier version of the
+              // code. This existing behavior of always incrementing the epoch following any
+              // successful shuffle map stage completion may have benefits by causing unneeded
+              // cached map outputs to be cleaned up earlier on executors. In the future we can
+              // consider removing this call, but this will require some extra investigation.
+              // See https://github.com/apache/spark/pull/17955/files#r117385673 for more details.
+              mapOutputTracker.incrementEpoch()
+
               clearCacheLocs()
 
               if (!shuffleStage.isAvailable) {