SPARK-52124 Actively Releasing Disk Space After Application Completio…

core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala

@@ -110,6 +110,9 @@ private[deploy] class Worker(
   private val PROLONGED_REGISTRATION_RETRY_INTERVAL_SECONDS = (math.round(60
     * REGISTRATION_RETRY_FUZZ_MULTIPLIER))
 
+  private val CLEANUP_WORKDIR_AFTER_FINISHED_APP_ENABLED =
+    conf.get(WORKER_CLEANUP_WORKDIR_AFTER_FINISHED_APP_ENABLED)
+
   private val CLEANUP_ENABLED = conf.get(WORKER_CLEANUP_ENABLED)
   // How often worker will clean up old app folders
   private val CLEANUP_INTERVAL_MILLIS = conf.get(WORKER_CLEANUP_INTERVAL) * 1000
@@ -694,6 +697,9 @@ private[deploy] class Worker(
     case ApplicationFinished(id) =>
       finishedApps += id
       maybeCleanupApplication(id)
+      if (CLEANUP_WORKDIR_AFTER_FINISHED_APP_ENABLED) {
+        maybeCleanupApplicationWorkDir(id)
+      }
 
     case DecommissionWorker =>
       decommissionSelf()
@@ -728,6 +734,32 @@ private[deploy] class Worker(
     registerWithMaster()
   }
 
+  private def maybeCleanupApplicationWorkDir(id: String): Unit = {
+    val appIds = (executors.values.map(_.appId) ++ drivers.values.map(_.driverId)).toSet
+    try {
+      val cleanupFuture: concurrent.Future[Unit] = concurrent.Future {
+        val appDir = new File(workDir, id)
+        if (appDir.isDirectory && appDir.exists()) {
+          val isAppStillRunning = appIds.contains(id)
+          if (!isAppStillRunning) {
+            logInfo(s"Removing application work directory: ${appDir.getPath}")
+            Utils.deleteRecursively(appDir)
+            if (conf.get(config.SHUFFLE_SERVICE_DB_ENABLED) &&
+              conf.get(config.SHUFFLE_SERVICE_ENABLED)) {
+              shuffleService.applicationRemoved(appDir.getName)
+            }
+          }
+        }
+      }(cleanupThreadExecutor)
+      cleanupFuture.failed.foreach(e =>
+        logError("App dir cleanup failed: " + e.getMessage, e)
+      )(cleanupThreadExecutor)
+    } catch {
+      case _: RejectedExecutionException if cleanupThreadExecutor.isShutdown =>
+        logWarning("Failed to cleanup work dir as executor pool was shutdown")
+    }
+  }
+
   private def maybeCleanupApplication(id: String): Unit = {
     val shouldCleanup = finishedApps.contains(id) && !executors.values.exists(_.appId == id)
     if (shouldCleanup) {
@@ -914,6 +946,9 @@ private[deploy] class Worker(
             exitStatus.map(" exitStatus " + _).getOrElse(""))
       }
       maybeCleanupApplication(appId)
+      if (CLEANUP_WORKDIR_AFTER_FINISHED_APP_ENABLED) {
+        maybeCleanupApplicationWorkDir(appId)
+      }
     }
   }
 }

core/src/main/scala/org/apache/spark/internal/config/Worker.scala

@@ -42,6 +42,12 @@ private[spark] object Worker {
     .timeConf(TimeUnit.MILLISECONDS)
     .createWithDefaultString("10s")
 
+  val WORKER_CLEANUP_WORKDIR_AFTER_FINISHED_APP_ENABLED =
+    ConfigBuilder("spark.worker.cleanupWorkdirAfterAppFinished.enabled")
+    .version("1.0.0")
+    .booleanConf
+    .createWithDefault(false)
+
   val WORKER_CLEANUP_ENABLED = ConfigBuilder("spark.worker.cleanup.enabled")
     .version("1.0.0")
     .booleanConf