From 6ce6cba0d2469630ebfa9b53782c1018ee6ea873 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:24:04 -0700 Subject: [PATCH 1/3] feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse deleteWarehouseEmptyFirst drops every table with purgeRequested=true, then immediately deletes the warehouse entity. Lakekeeper purges the dropped tables' data files asynchronously (task queue `tabular_purge`) and refuses to delete the warehouse while any purge is pending -- the tasks need the warehouse's storage profile to reach S3, so deleting it first would orphan them and leak the files. It answers 409 WarehouseHasUnfinishedTasks until the queue drains, so the first delete of any warehouse that had ever stored execution results always failed; a retry seconds later succeeded. Retry the final warehouse delete on exactly that conflict -- 409 with error.type WarehouseHasUnfinishedTasks -- with a bounded pause (default 10 retries x 2s; the queue normally drains within seconds). Every other error, including any other 409, still fails immediately, and 404 stays the idempotent goal state. The bound and delay are constructor parameters with defaults, so production call sites are unchanged and the spec injects a zero delay -- no real sleeps in the tests. LakekeeperClientSpec covers the three outcomes against its in-process stub: 409-409-204 succeeds with exactly 3 attempts (failed before the fix on the first 409), a never-draining queue fails after the bounded 1+3 attempts, and a 409 of any other type fails on the first attempt with no retry. Found while testing the flag-gated per-user warehouse feature; no deployment is affected because the flag defaults to off. Closes #7742. --- .../texera/web/service/LakekeeperClient.scala | 46 +++++++++++-- .../web/service/LakekeeperClientSpec.scala | 67 +++++++++++++++++++ 2 files changed, 109 insertions(+), 4 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala index 7fdafc0df71..dc221750773 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala @@ -39,8 +39,19 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala * * @param catalogUri the Iceberg REST catalog uri (ends with `/catalog`), from which the * management base is derived. Overridable for tests. + * @param unfinishedTasksRetries how many times the final warehouse delete is retried while + * Lakekeeper reports 409 WarehouseHasUnfinishedTasks — its + * asynchronous purge of the dropped tables' data files is + * still draining (#7742). With the default delay this bounds + * the wait at ~20s; the queue normally drains within seconds. + * @param unfinishedTasksRetryDelayMillis pause between those retries. Overridable for tests + * (0 keeps the spec free of real sleeps). */ -class LakekeeperClient(catalogUri: String = StorageConfig.icebergRESTCatalogUri) { +class LakekeeperClient( + catalogUri: String = StorageConfig.icebergRESTCatalogUri, + unfinishedTasksRetries: Int = 10, + unfinishedTasksRetryDelayMillis: Long = 2000 +) { // Lakekeeper's default project; single-project deployments (ours) use the nil UUID. private val DefaultProjectId = "00000000-0000-0000-0000-000000000000" @@ -126,12 +137,39 @@ class LakekeeperClient(catalogUri: String = StorageConfig.icebergRESTCatalogUri) failOn(response.getStatus, response.getBody, s"drop namespace '$namespace'") } } - val response = Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString() - if (response.getStatus != 404) { - failOn(response.getStatus, response.getBody, "delete warehouse") + // The drops above purge each table's data files asynchronously (Lakekeeper task + // queue `tabular_purge`), and Lakekeeper refuses to delete the warehouse while + // any purge is pending — the tasks need the warehouse's storage profile to reach + // S3, so deleting it first would orphan them and leak the files. It answers 409 + // WarehouseHasUnfinishedTasks until the queue drains (normally within seconds), + // so ride that out with a bounded retry; every other error, including any other + // 409, still fails immediately. (#7742) + var attempt = 0 + var deleted = false + while (!deleted) { + val response = Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString() + attempt += 1 + if (response.getStatus == 404 || (response.getStatus >= 200 && response.getStatus < 300)) { + deleted = true + } else if ( + isUnfinishedTasksConflict(response.getStatus, response.getBody) && + attempt <= unfinishedTasksRetries + ) { + Thread.sleep(unfinishedTasksRetryDelayMillis) + } else { + failOn(response.getStatus, response.getBody, "delete warehouse") + } } } + /** Lakekeeper's "purge queue still draining" conflict — the only retried error. */ + private def isUnfinishedTasksConflict(status: Int, body: String): Boolean = + status == 409 && (try { + mapper.readTree(body).path("error").path("type").asText() == "WarehouseHasUnfinishedTasks" + } catch { + case _: Exception => false + }) + /** Top-level namespaces in the warehouse. Texera's execution namespaces are single-level. */ private def listNamespaces(warehouseId: UUID): List[String] = fetchAllPages(s"$catalogBase/$warehouseId/namespaces", "namespaces", "list namespaces")(parts => diff --git a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala index ac047c590c5..ed0aeefb041 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala @@ -65,13 +65,41 @@ class LakekeeperClientSpec exchange.close() } + // Lakekeeper purges dropped tables asynchronously (queue `tabular_purge`), and + // answers a warehouse delete with 409 WarehouseHasUnfinishedTasks while any + // purge task is pending (#7742). These stub warehouses model that queue: + // `racing` drains after two attempts, `alwaysBusy` never drains, and + // `otherConflict` 409s for an unrelated reason (which must NOT be retried). + private val racingWarehouseId = UUID.randomUUID() + private val alwaysBusyWarehouseId = UUID.randomUUID() + private val otherConflictWarehouseId = UUID.randomUUID() + @volatile private var racingDeleteAttempts = 0 + @volatile private var busyDeleteAttempts = 0 + private val unfinishedTasksBody = + """{"error":{"message":"Warehouse has unfinished tasks. Cannot delete warehouse until all tasks are finished.","type":"WarehouseHasUnfinishedTasks","code":409}}""" + server.createContext( "/management/v1/warehouse", (exchange: HttpExchange) => { record(exchange) + val path = exchange.getRequestURI.getPath + val isDelete = exchange.getRequestMethod == "DELETE" if (exchange.getRequestMethod == "POST") { lastCreateBody = new String(exchange.getRequestBody.readAllBytes(), StandardCharsets.UTF_8) respond(exchange, 201, s"""{"warehouse-id": "$warehouseId"}""") + } else if (isDelete && path.endsWith(racingWarehouseId.toString)) { + racingDeleteAttempts += 1 + if (racingDeleteAttempts <= 2) respond(exchange, 409, unfinishedTasksBody) + else respond(exchange, 204, "") + } else if (isDelete && path.endsWith(alwaysBusyWarehouseId.toString)) { + busyDeleteAttempts += 1 + respond(exchange, 409, unfinishedTasksBody) + } else if (isDelete && path.endsWith(otherConflictWarehouseId.toString)) { + respond( + exchange, + 409, + """{"error":{"message":"warehouse is in use","type":"Conflict","code":409}}""" + ) } else { respond(exchange, 200, "{}") } @@ -121,9 +149,19 @@ class LakekeeperClientSpec s"http://localhost:${server.getAddress.getPort}/catalog" ) + // Zero retry delay keeps the spec free of real sleeps (deterministic); 3 + // retries keeps the exhaustion case cheap to assert. + private val retryClient = new LakekeeperClient( + s"http://localhost:${server.getAddress.getPort}/catalog", + unfinishedTasksRetries = 3, + unfinishedTasksRetryDelayMillis = 0 + ) + override protected def beforeEach(): Unit = { requests.synchronized { requests.clear() } lastCreateBody = "" + racingDeleteAttempts = 0 + busyDeleteAttempts = 0 } override protected def afterAll(): Unit = server.stop(0) @@ -168,4 +206,33 @@ class LakekeeperClientSpec error.getMessage should include("Lakekeeper") error.getMessage should include("500") } + + it should "wait out 409 WarehouseHasUnfinishedTasks from the asynchronous purge (#7742)" in { + // Lakekeeper purges dropped tables asynchronously; the stub answers the + // warehouse delete with 409 WarehouseHasUnfinishedTasks twice before the + // queue "drains" and it returns 204. The delete must ride that out. + noException should be thrownBy retryClient.deleteWarehouseEmptyFirst(racingWarehouseId) + racingDeleteAttempts shouldBe 3 + } + + it should "give up once the purge-wait retries are exhausted" in { + val error = intercept[RuntimeException] { + retryClient.deleteWarehouseEmptyFirst(alwaysBusyWarehouseId) + } + error.getMessage should include("409") + error.getMessage should include("WarehouseHasUnfinishedTasks") + // 1 initial attempt + 3 retries, then fail -- the wait is bounded. + busyDeleteAttempts shouldBe 4 + } + + it should "fail immediately on a 409 that is not WarehouseHasUnfinishedTasks" in { + val error = intercept[RuntimeException] { + retryClient.deleteWarehouseEmptyFirst(otherConflictWarehouseId) + } + error.getMessage should include("409") + val deletes = requests.synchronized { + requests.count(_ == s"DELETE /management/v1/warehouse/$otherConflictWarehouseId") + } + deletes shouldBe 1 + } } From 3c3a96773f91b759fac72e9e8439e1734576c594 Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:53:11 -0700 Subject: [PATCH 2/3] test(amber): pin the malformed-409 path in the purge-wait retry The retry only fires for a 409 whose body carries the WarehouseHasUnfinishedTasks type, and the type check parses that body -- so a 409 with a non-JSON body (a gateway error page, say) must read as 'not the purge conflict' and fail immediately rather than be waited out as if it were transient. Assert it fails on the first attempt. --- .../web/service/LakekeeperClientSpec.scala | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala index ed0aeefb041..413aedccd66 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala @@ -73,6 +73,8 @@ class LakekeeperClientSpec private val racingWarehouseId = UUID.randomUUID() private val alwaysBusyWarehouseId = UUID.randomUUID() private val otherConflictWarehouseId = UUID.randomUUID() + private val malformedConflictWarehouseId = UUID.randomUUID() + @volatile private var malformedDeleteAttempts = 0 @volatile private var racingDeleteAttempts = 0 @volatile private var busyDeleteAttempts = 0 private val unfinishedTasksBody = @@ -94,6 +96,10 @@ class LakekeeperClientSpec } else if (isDelete && path.endsWith(alwaysBusyWarehouseId.toString)) { busyDeleteAttempts += 1 respond(exchange, 409, unfinishedTasksBody) + } else if (isDelete && path.endsWith(malformedConflictWarehouseId.toString)) { + // A 409 whose body isn't the JSON envelope the type check reads. + malformedDeleteAttempts += 1 + respond(exchange, 409, "gateway conflict") } else if (isDelete && path.endsWith(otherConflictWarehouseId.toString)) { respond( exchange, @@ -162,6 +168,7 @@ class LakekeeperClientSpec lastCreateBody = "" racingDeleteAttempts = 0 busyDeleteAttempts = 0 + malformedDeleteAttempts = 0 } override protected def afterAll(): Unit = server.stop(0) @@ -225,6 +232,16 @@ class LakekeeperClientSpec busyDeleteAttempts shouldBe 4 } + it should "fail immediately on a 409 whose body is not the expected JSON envelope" in { + // The type check parses the body; a malformed one must read as "not the + // purge conflict" and fail rather than be retried as if it were transient. + val error = intercept[RuntimeException] { + retryClient.deleteWarehouseEmptyFirst(malformedConflictWarehouseId) + } + error.getMessage should include("409") + malformedDeleteAttempts shouldBe 1 + } + it should "fail immediately on a 409 that is not WarehouseHasUnfinishedTasks" in { val error = intercept[RuntimeException] { retryClient.deleteWarehouseEmptyFirst(otherConflictWarehouseId) From 9fc3dd4b13c238e4c7950c2b07f15985b5c78b6b Mon Sep 17 00:00:00 2001 From: mengw15 <125719918+mengw15@users.noreply.github.com> Date: Mon, 17 Aug 2026 17:55:50 -0700 Subject: [PATCH 3/3] refactor(amber): back off exponentially while waiting out the purge queue The wait was a fixed 2s x 10. That made the common case -- a purge that drains almost immediately -- cost the caller a full 2s before the delete was retried, on a request a user is waiting on, while still issuing 11 requests when the queue is genuinely slow. Double the pause instead, from 200ms up to a 5s cap over 7 retries: 0.2+0.4+0.8+1.6+3.2+5+5s, so a fast purge returns in ~200ms, a slow one issues fewer requests, and the overall bound drops from 20s to ~16s. Tests still inject a zero initial delay -- doubling zero stays zero, so they contain no real sleeps. --- .../texera/web/service/LakekeeperClient.scala | 23 +++++++++++++------ .../web/service/LakekeeperClientSpec.scala | 2 +- 2 files changed, 17 insertions(+), 8 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala index dc221750773..0f0a909e3ad 100644 --- a/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala +++ b/amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala @@ -42,15 +42,22 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala * @param unfinishedTasksRetries how many times the final warehouse delete is retried while * Lakekeeper reports 409 WarehouseHasUnfinishedTasks — its * asynchronous purge of the dropped tables' data files is - * still draining (#7742). With the default delay this bounds - * the wait at ~20s; the queue normally drains within seconds. - * @param unfinishedTasksRetryDelayMillis pause between those retries. Overridable for tests - * (0 keeps the spec free of real sleeps). + * still draining (#7742). + * @param unfinishedTasksInitialDelayMillis first pause between those retries; it doubles up + * to the cap. Starting small keeps a fast purge + * (the common case) from costing the caller a full + * fixed interval, while the growth keeps a slow one + * from hammering Lakekeeper. Overridable for tests + * (0 keeps the spec free of real sleeps — doubling + * 0 stays 0). + * @param unfinishedTasksMaxDelayMillis cap for that doubling. With the defaults the retries + * wait 0.2+0.4+0.8+1.6+3.2+5+5s ≈ 16s in total. */ class LakekeeperClient( catalogUri: String = StorageConfig.icebergRESTCatalogUri, - unfinishedTasksRetries: Int = 10, - unfinishedTasksRetryDelayMillis: Long = 2000 + unfinishedTasksRetries: Int = 7, + unfinishedTasksInitialDelayMillis: Long = 200, + unfinishedTasksMaxDelayMillis: Long = 5000 ) { // Lakekeeper's default project; single-project deployments (ours) use the nil UUID. @@ -145,6 +152,7 @@ class LakekeeperClient( // so ride that out with a bounded retry; every other error, including any other // 409, still fails immediately. (#7742) var attempt = 0 + var delay = unfinishedTasksInitialDelayMillis var deleted = false while (!deleted) { val response = Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString() @@ -155,7 +163,8 @@ class LakekeeperClient( isUnfinishedTasksConflict(response.getStatus, response.getBody) && attempt <= unfinishedTasksRetries ) { - Thread.sleep(unfinishedTasksRetryDelayMillis) + Thread.sleep(delay) + delay = math.min(delay * 2, unfinishedTasksMaxDelayMillis) } else { failOn(response.getStatus, response.getBody, "delete warehouse") } diff --git a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala index 413aedccd66..09fe0fd6a19 100644 --- a/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala @@ -160,7 +160,7 @@ class LakekeeperClientSpec private val retryClient = new LakekeeperClient( s"http://localhost:${server.getAddress.getPort}/catalog", unfinishedTasksRetries = 3, - unfinishedTasksRetryDelayMillis = 0 + unfinishedTasksInitialDelayMillis = 0 ) override protected def beforeEach(): Unit = {