diff --git a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala b/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala index 0f29d19ecc846..e7163cd695f1b 100644 --- a/core/src/main/scala/org/apache/spark/deploy/history/HistoryServerDiskManager.scala +++ b/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) } diff --git a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala index f9ab2de28c591..b1cfc75f85b9d 100644 --- a/core/src/test/scala/org/apache/spark/deploy/history/HistoryServerDiskManagerSuite.scala +++ b/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())