S3ShuffleManager: Only try to clean up shuffle files if there are no …

…registered shuffle ids.

Signed-off-by: Pascal Spörri <[email protected]>

src/main/scala/org/apache/spark/shuffle/sort/S3ShuffleManager.scala

@@ -167,16 +167,19 @@ private[spark] class S3ShuffleManager(conf: SparkConf) extends ShuffleManager wi
 
   /** Shut down this ShuffleManager. */
   override def stop(): Unit = {
+    val cleanupRequired = registeredShuffleIds.size > 0
     registeredShuffleIds.foreach(
       shuffleId => {
         purgeCaches(shuffleId)
         registeredShuffleIds.remove(shuffleId)
       })
-    if (dispatcher.cleanupShuffleFiles) {
-      logInfo(f"Cleaning up shuffle files in ${dispatcher.rootDir}.")
-      dispatcher.removeRoot()
-    } else {
-      logInfo(f"Manually cleanup shuffle files in ${dispatcher.rootDir}")
+    if (cleanupRequired) {
+      if (dispatcher.cleanupShuffleFiles) {
+        logInfo(f"Cleaning up shuffle files in ${dispatcher.rootDir}.")
+        dispatcher.removeRoot()
+      } else {
+        logInfo(f"Manually cleanup shuffle files in ${dispatcher.rootDir}")
+      }
     }
     shuffleBlockResolver.stop()
   }