You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2019/03/14 08:19:50 UTC
[GitHub] [spark] ajithme commented on a change in pull request #24056:
[SPARK-26152] Synchronize Worker Cleanup with Worker Shutdown
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_r265458000
##########
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:
Yes. https://github.com/apache/spark/pull/24056#issuecomment-4715890
Now this JIRA will make cleanup failures to be skipped gracefully instead of throwing exception
----------------------------------------------------------------
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:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org