From c18a5515015a9405b9a08c8ad225073fd0088538 Mon Sep 17 00:00:00 2001 From: Uros Bojanic <221401595+uros-b@users.noreply.github.com> Date: Fri, 11 Sep 2026 13:11:28 +0000 Subject: [PATCH 1/2] [SPARK-59439][CORE] Release the HistoryServerDiskManager lease reservation exactly once on a failed commit Lease.commit() releases the reservation before the fallible rename; when the rename fails and the caller rolls back, the reservation is released a second time, driving the usage tracker negative. Funnel both commit() and rollback() through an idempotent releaseLease() so it is returned exactly once. --- .../history/HistoryServerDiskManager.scala | 17 ++++++++-- .../HistoryServerDiskManagerSuite.scala | 31 ++++++++++++++++++- 2 files changed, 45 insertions(+), 3 deletions(-) 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..d26eea4b1fb42 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,35 @@ 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 reservation was returned exactly once: usage is back to zero (not negative) and the + // full capacity is free again. + assert(manager.committed() === 0) + 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()) From 4cfb8786bd171dd672d158286b95f6e59b2034d3 Mon Sep 17 00:00:00 2001 From: Uros Bojanic <221401595+uros-b@users.noreply.github.com> Date: Sat, 12 Sep 2026 07:26:54 +0000 Subject: [PATCH 2/2] Address review --- .../spark/deploy/history/HistoryServerDiskManagerSuite.scala | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 d26eea4b1fb42..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 @@ -427,9 +427,8 @@ abstract class HistoryServerDiskManagerSuite extends SparkFunSuite with BeforeAn // The caller rolls the lease back after the failed commit, as FsHistoryProvider does. lease.rollback() - // The reservation was returned exactly once: usage is back to zero (not negative) and the - // full capacity is free again. - assert(manager.committed() === 0) + // 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.