feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse

amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala

@@ -39,8 +39,26 @@ 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).
+  * @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) {
+class LakekeeperClient(
+    catalogUri: String = StorageConfig.icebergRESTCatalogUri,
+    unfinishedTasksRetries: Int = 7,
+    unfinishedTasksInitialDelayMillis: Long = 200,
+    unfinishedTasksMaxDelayMillis: Long = 5000
+) {
 
   // Lakekeeper's default project; single-project deployments (ours) use the nil UUID.
   private val DefaultProjectId = "00000000-0000-0000-0000-000000000000"
@@ -126,12 +144,41 @@ 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 delay = unfinishedTasksInitialDelayMillis
+    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(delay)
+        delay = math.min(delay * 2, unfinishedTasksMaxDelayMillis)
+      } 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 =>

amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala

@@ -65,13 +65,47 @@ 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()
+  private val malformedConflictWarehouseId = UUID.randomUUID()
+  @volatile private var malformedDeleteAttempts = 0
+  @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(malformedConflictWarehouseId.toString)) {
+        // A 409 whose body isn't the JSON envelope the type check reads.
+        malformedDeleteAttempts += 1
+        respond(exchange, 409, "<html>gateway conflict</html>")
+      } 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 +155,20 @@ 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,
+    unfinishedTasksInitialDelayMillis = 0
+  )
+
   override protected def beforeEach(): Unit = {
     requests.synchronized { requests.clear() }
     lastCreateBody = ""
+    racingDeleteAttempts = 0
+    busyDeleteAttempts = 0
+    malformedDeleteAttempts = 0
   }
 
   override protected def afterAll(): Unit = server.stop(0)
@@ -168,4 +213,43 @@ 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 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)
+    }
+    error.getMessage should include("409")
+    val deletes = requests.synchronized {
+      requests.count(_ == s"DELETE /management/v1/warehouse/$otherConflictWarehouseId")
+    }
+    deletes shouldBe 1
+  }
 }