ajithme commented on a change in pull request #24056: [SPARK-26152] Synchronize
Worker Cleanup with Worker Shutdown
URL: https://github.com/apache/spark/pull/24056#discussion_r265117091
##########
File path: core/src/main/scala/org/apache/spark/deploy/worker/Worker.scala
##########
@@ -450,27 +450,32 @@ private[deploy] class Worker(
// rpcEndpoint.
// Copy ids so that it can be used in the cleanup thread.
val appIds = (executors.values.map(_.appId) ++
drivers.values.map(_.driverId)).toSet
- val cleanupFuture = concurrent.Future {
- val appDirs = workDir.listFiles()
- if (appDirs == null) {
- throw new IOException("ERROR: Failed to list files in " + appDirs)
- }
- appDirs.filter { dir =>
- // the directory is used by an application - check that the
application is not running
- // when cleaning up
- val appIdFromDir = dir.getName
- val isAppStillRunning = appIds.contains(appIdFromDir)
- dir.isDirectory && !isAppStillRunning &&
- !Utils.doesDirectoryContainAnyNewFiles(dir,
APP_DATA_RETENTION_SECONDS)
- }.foreach { dir =>
- logInfo(s"Removing directory: ${dir.getPath}")
- Utils.deleteRecursively(dir)
- }
- }(cleanupThreadExecutor)
+ try {
+ val cleanupFuture: concurrent.Future[Unit] = concurrent.Future {
+ val appDirs = workDir.listFiles()
+ if (appDirs == null) {
+ throw new IOException("ERROR: Failed to list files in " + appDirs)
+ }
+ appDirs.filter { dir =>
+ // the directory is used by an application - check that the
application is not running
+ // when cleaning up
+ val appIdFromDir = dir.getName
+ val isAppStillRunning = appIds.contains(appIdFromDir)
+ dir.isDirectory && !isAppStillRunning &&
+ !Utils.doesDirectoryContainAnyNewFiles(dir,
APP_DATA_RETENTION_SECONDS)
+ }.foreach { dir =>
+ logInfo(s"Removing directory: ${dir.getPath}")
+ Utils.deleteRecursively(dir)
+ }
+ }(cleanupThreadExecutor)
- cleanupFuture.failed.foreach(e =>
- logError("App dir cleanup failed: " + e.getMessage, e)
- )(cleanupThreadExecutor)
+ cleanupFuture.failed.foreach(e =>
+ logError("App dir cleanup failed: " + e.getMessage, e)
+ )(cleanupThreadExecutor)
+ } catch {
+ case _ : RejectedExecutionException if
cleanupThreadExecutor.isShutdown =>
Review comment:
I have updated accordingly. please review
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
