[SPARK-59439][CORE] Release the HistoryServerDiskManager lease reservation exactly once on a failed commit

core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala

@@ -332,6 +332,19 @@ private class HistoryServerDiskManager(
 
   private[history] class Lease(val tmpPath: File, private val leased: Long) {
 
+    // The leased (reserved, uncommitted) usage must be returned exactly once, whether the lease
+    // is committed or rolled back. commit() releases it before moving the store into place, so a
+    // failure after that point (e.g. a failed rename) sends the caller through rollback(); guard
+    // against releasing it a second time there, which would drive the usage tracker negative.
+    private var released = false
+
+    private def releaseLease(): Unit = {
+      if (!released) {
+        updateUsage(-leased)
+        released = true
+      }
+    }
+
     /**
      * Commits a lease to its final location, and update accounting information. This method
      * marks the application as active, so its store is not available for eviction.
@@ -357,7 +370,7 @@ private class HistoryServerDiskManager(
         }
       }
 
-      updateUsage(-leased)
+      releaseLease()
 
       val newSize = sizeOf(tmpPath)
       makeRoom(newSize)
@@ -385,7 +398,7 @@ private class HistoryServerDiskManager(
 
     /** Deletes the temporary directory created for the lease. */
     def rollback(): Unit = {
-      updateUsage(-leased)
+      releaseLease()
       Utils.deleteRecursively(tmpPath)
     }
 

core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy.history
 
-import java.io.File
+import java.io.{File, IOException}
 import java.util.concurrent.{CountDownLatch, FutureTask, TimeUnit}
 
 import scala.concurrent.duration._
@@ -411,6 +411,34 @@ abstract class HistoryServerDiskManagerSuite extends SparkFunSuite with BeforeAn
     assert(manager.committed() === 0)
   }
 
+  test("SPARK-59439: a failed commit rename does not double-release the lease") {
+    val manager = mockManager()
+
+    // Reserve space for a store, then make the rename in commit() fail by removing the source
+    // directory. commit() releases the lease reservation before the rename, so a failure there
+    // must not let the caller's rollback() deduct the reservation a second time.
+    val lease = manager.lease(2)
+    doReturn(2L).when(manager).sizeOf(meq(lease.tmpPath))
+    Utils.deleteRecursively(lease.tmpPath)
+
+    intercept[IOException] {
+      lease.commit("app1", None)
+    }
+    // The caller rolls the lease back after the failed commit, as FsHistoryProvider does.
+    lease.rollback()
+
+    // The leased reservation was returned exactly once: the current (leased) usage is back to
+    // zero, not negative, so the full capacity is free again. committed() is untouched here.
+    assert(manager.free() === MAX_USAGE)
+
+    // Accounting is intact, so a subsequent lease and commit still succeed.
+    val lease2 = manager.lease(2)
+    doReturn(2L).when(manager).sizeOf(meq(lease2.tmpPath))
+    val dst = lease2.commit("app1", None)
+    assert(dst.isDirectory())
+    assert(manager.committed() === 2)
+  }
+
   test("SPARK-38095: appStorePath should use backend extensions") {
     val conf = new SparkConf().set(HYBRID_STORE_DISK_BACKEND, backend.toString)
     val manager = new HistoryServerDiskManager(conf, testDir, store, new ManualClock())