diff --git a/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala b/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
index a7ed9226c574a..7c7670b5095e2 100644
--- a/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala
@@ -250,6 +250,8 @@ private[spark] class DiskBlockManager(
     Utils.getConfiguredLocalDirs(conf).flatMap { rootDir =>
       try {
         val localDir = Utils.createDirectory(rootDir, "blockmgr")
+        // SPARK-41946: to clearup remained dir.
+        ShutdownHookManager.registerShutdownDeleteDir(localDir)
         logInfo(s"Created local directory at $localDir")
         Some(localDir)
       } catch {