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..ed95d8e4f40 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 @@ -20,8 +20,11 @@ package org.apache.texera.web.service import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper} +import com.typesafe.scalalogging.LazyLogging import kong.unirest.Unirest import org.apache.texera.common.config.StorageConfig +import org.apache.texera.common.util.RetryUtil +import org.apache.texera.web.service.LakekeeperClient.PurgeWaitPolicy import java.net.URLEncoder import java.nio.charset.StandardCharsets @@ -29,6 +32,25 @@ import java.util.UUID import scala.collection.mutable.ListBuffer import scala.jdk.CollectionConverters.IteratorHasAsScala +object LakekeeperClient { + + /** + * How long the final warehouse delete waits out Lakekeeper's asynchronous purge of the + * dropped tables' data files, which it reports as 409 WarehouseHasUnfinishedTasks while + * still draining (#7742). The pause starts at `initialDelayMillis` and doubles up to + * `maxDelayMillis`: 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. With the defaults the waits between the `maxAttempts` attempts sum to + * 0.2+0.4+0.8+1.6+3.2+5+5s ≈ 16s. Overridable for tests (a 0 initial delay keeps the + * spec free of real sleeps — doubling 0 stays 0). + */ + final case class PurgeWaitPolicy( + maxAttempts: Int = 8, + initialDelayMillis: Long = 200, + maxDelayMillis: Long = 5000 + ) +} + /** * Client for the Lakekeeper APIs used to manage per-user warehouses (#6870). * @@ -39,8 +61,13 @@ 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 purgeWait how long the final warehouse delete waits out Lakekeeper's asynchronous + * purge of the dropped tables' data files (#7742). */ -class LakekeeperClient(catalogUri: String = StorageConfig.icebergRESTCatalogUri) { +class LakekeeperClient( + catalogUri: String = StorageConfig.icebergRESTCatalogUri, + purgeWait: PurgeWaitPolicy = PurgeWaitPolicy() +) extends LazyLogging { // Lakekeeper's default project; single-project deployments (ours) use the nil UUID. private val DefaultProjectId = "00000000-0000-0000-0000-000000000000" @@ -126,12 +153,45 @@ 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) + RetryUtil.withBackoff( + description = DeleteWarehouseAction, + maxAttempts = purgeWait.maxAttempts, + initialDelayMillis = purgeWait.initialDelayMillis, + onRetry = attempt => logger.warn(attempt.message), + maxDelayMillis = purgeWait.maxDelayMillis, + shouldRetry = _.isInstanceOf[UnfinishedTasksConflictException] + ) { + val response = Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString() + response.getStatus match { + case 404 => // already gone — the idempotent goal state + case 409 if isUnfinishedTasksBody(response.getBody) => + throw new UnfinishedTasksConflictException(response.getBody) + case status => failOn(status, response.getBody, DeleteWarehouseAction) + } } } + private val DeleteWarehouseAction = "delete warehouse" + + /** Tags the one retryable delete failure so the backoff predicate can single it out. */ + private class UnfinishedTasksConflictException(body: String) + extends RuntimeException(s"Lakekeeper $DeleteWarehouseAction failed (HTTP 409): $body") + + /** Lakekeeper's "purge queue still draining" 409 body — the only retried error. */ + private def isUnfinishedTasksBody(body: String): Boolean = + 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..1451720eb04 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 @@ -21,6 +21,7 @@ package org.apache.texera.web.service import com.fasterxml.jackson.databind.ObjectMapper import com.sun.net.httpserver.{HttpExchange, HttpServer} +import org.apache.texera.web.service.LakekeeperClient.PurgeWaitPolicy import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} @@ -57,6 +58,11 @@ class LakekeeperClientSpec line } + // How many warehouse deletes have reached the stub for this id, including the one + // being served: `record` logs every request before the handler dispatches on it. + private def deleteAttempts(id: UUID): Int = + requests.synchronized { requests.count(_ == s"DELETE /management/v1/warehouse/$id") } + private def respond(exchange: HttpExchange, status: Int, body: String): Unit = { val bytes = body.getBytes(StandardCharsets.UTF_8) exchange.getResponseHeaders.add("Content-Type", "application/json") @@ -65,15 +71,43 @@ 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() + 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) - if (exchange.getRequestMethod == "POST") { - lastCreateBody = new String(exchange.getRequestBody.readAllBytes(), StandardCharsets.UTF_8) - respond(exchange, 201, s"""{"warehouse-id": "$warehouseId"}""") - } else { - respond(exchange, 200, "{}") + (exchange.getRequestMethod, exchange.getRequestURI.getPath) match { + case ("POST", _) => + lastCreateBody = + new String(exchange.getRequestBody.readAllBytes(), StandardCharsets.UTF_8) + respond(exchange, 201, s"""{"warehouse-id": "$warehouseId"}""") + case ("DELETE", path) if path.endsWith(racingWarehouseId.toString) => + if (deleteAttempts(racingWarehouseId) <= 2) respond(exchange, 409, unfinishedTasksBody) + else respond(exchange, 204, "") + case ("DELETE", path) if path.endsWith(alwaysBusyWarehouseId.toString) => + respond(exchange, 409, unfinishedTasksBody) + case ("DELETE", path) if path.endsWith(malformedConflictWarehouseId.toString) => + // A 409 whose body isn't the JSON envelope the type check reads. + respond(exchange, 409, "gateway conflict") + case ("DELETE", path) if path.endsWith(otherConflictWarehouseId.toString) => + respond( + exchange, + 409, + """{"error":{"message":"warehouse is in use","type":"Conflict","code":409}}""" + ) + case _ => + respond(exchange, 200, "{}") } } ) @@ -117,8 +151,15 @@ class LakekeeperClientSpec ) server.start() - private val client = new LakekeeperClient( - s"http://localhost:${server.getAddress.getPort}/catalog" + private val stubCatalogUri = s"http://localhost:${server.getAddress.getPort}/catalog" + + private val client = new LakekeeperClient(stubCatalogUri) + + // Zero retry delay keeps the spec free of real sleeps (deterministic); 4 + // attempts keeps the exhaustion case cheap to assert. + private val retryClient = new LakekeeperClient( + stubCatalogUri, + PurgeWaitPolicy(maxAttempts = 4, initialDelayMillis = 0) ) override protected def beforeEach(): Unit = { @@ -168,4 +209,41 @@ 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) + deleteAttempts(racingWarehouseId) 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") + deleteAttempts(alwaysBusyWarehouseId) shouldBe 4 + } + + // Shared by the non-retryable-409 cases: the delete must fail on the first + // attempt, with the status surfaced, rather than be waited out as transient. + private def assertFailsWithoutRetry(id: UUID): Unit = { + val error = intercept[RuntimeException] { + retryClient.deleteWarehouseEmptyFirst(id) + } + error.getMessage should include("409") + deleteAttempts(id) shouldBe 1 + } + + 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. + assertFailsWithoutRetry(malformedConflictWarehouseId) + } + + it should "fail immediately on a 409 that is not WarehouseHasUnfinishedTasks" in { + assertFailsWithoutRetry(otherConflictWarehouseId) + } } diff --git a/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala index 3ebf094b4e7..63645a737c7 100644 --- a/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala +++ b/common/util/src/main/scala/org/apache/texera/common/util/RetryUtil.scala @@ -51,14 +51,17 @@ object RetryUtil { } /** - * Runs `operation`, retrying on failure with exponential backoff (the delay doubles after each - * failed attempt) until it succeeds or `maxAttempts` is reached. The final failure is wrapped - * with `description` and the last exception as its cause. + * Runs `operation`, retrying on failure with exponential backoff (the delay doubles after + * each failed attempt, capped at `maxDelayMillis`) until it succeeds or `maxAttempts` is + * reached. The final failure is wrapped with `description` and the last exception as its + * cause. * - * Only `NonFatal` failures are treated as transient, which is the same predicate the - * non-blocking sibling uses. Note that `NonFatal` admits non-fatal `Error`s -- `AssertionError`, - * `java.io.IOError`, `ServiceConfigurationError` -- so those are retried rather than propagated - * straight away. An `InterruptedException` -- raised by the operation or by the wait between + * Only `NonFatal` failures that `shouldRetry` accepts are treated as transient; a failure it + * rejects propagates immediately, unwrapped, spending no further attempts. The default accepts + * every `NonFatal` failure, which is the same predicate the non-blocking sibling uses. Note + * that `NonFatal` admits non-fatal `Error`s -- `AssertionError`, `java.io.IOError`, + * `ServiceConfigurationError` -- so those are retried rather than propagated straight away. + * An `InterruptedException` -- raised by the operation or by the wait between * attempts -- fails fast with the interrupt status restored, so a caller shutting the thread * down is never made to sit through the remaining backoff. * @@ -70,6 +73,11 @@ object RetryUtil { * caller's own logger, so retries are attributed to the caller rather * than to this util. * @param sleep how to wait; injectable so tests exercise the backoff without waiting. + * @param maxDelayMillis cap on any single wait: the doubling stops growing there, and an + * initial delay above it is clamped down. Unbounded by default. + * @param shouldRetry which failures are transient; the default retries every `NonFatal` + * one. A caller whose retry signal is response content rather than an + * exception type can throw a private marker and match it here. * @param operation the work to run, re-evaluated on each attempt. * @tparam T whatever `operation` returns. * @return `operation`'s value from the first attempt that succeeds. @@ -79,7 +87,9 @@ object RetryUtil { maxAttempts: Int, initialDelayMillis: Long, onRetry: RetryAttempt => Unit, - sleep: Long => Unit = Thread.sleep + sleep: Long => Unit = Thread.sleep, + maxDelayMillis: Long = Long.MaxValue, + shouldRetry: Throwable => Boolean = _ => true )(operation: => T): T = { // Restore the interrupt status and fail fast rather than retrying, whether the interrupt // arrives while running `operation` or while waiting between attempts. @@ -93,8 +103,8 @@ object RetryUtil { val outcome: Either[Throwable, T] = try Right(operation) catch { - case ie: InterruptedException => failInterrupted(ie) - case NonFatal(cause) => Left(cause) + case ie: InterruptedException => failInterrupted(ie) + case NonFatal(cause) if shouldRetry(cause) => Left(cause) } outcome match { @@ -109,10 +119,10 @@ object RetryUtil { onRetry(RetryAttempt(description, attempt, maxAttempts, delayMillis, cause)) try sleep(delayMillis) catch { case ie: InterruptedException => failInterrupted(ie) } - attemptFrom(attempt + 1, delayMillis * 2) + attemptFrom(attempt + 1, math.min(delayMillis * 2, maxDelayMillis)) } } - attemptFrom(attempt = 1, delayMillis = initialDelayMillis) + attemptFrom(attempt = 1, delayMillis = math.min(initialDelayMillis, maxDelayMillis)) } } diff --git a/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala index 828cb8d01e5..4ac630a0054 100644 --- a/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala +++ b/common/util/src/test/scala/org/apache/texera/common/util/RetryUtilSpec.scala @@ -29,9 +29,9 @@ import scala.util.control.ControlThrowable * Contract of the shared blocking backoff retry. `sleep` is injected everywhere so the backoff * progression is asserted exactly without any test waiting. * - * Coverage is the full contract both blocking callers rely on: the doubling progression, which - * failures count as transient, the give-up wrapping, and interrupt fail-fast during the operation - * and during a backoff sleep. + * Coverage is the full contract the blocking callers rely on: the doubling progression and its + * cap, which failures count as transient (the `NonFatal` gate and the `shouldRetry` predicate), + * the give-up wrapping, and interrupt fail-fast during the operation and during a backoff sleep. */ class RetryUtilSpec extends AnyFlatSpec { @@ -203,4 +203,71 @@ class RetryUtilSpec extends AnyFlatSpec { assert(delays.toList == List(200L, 400L)) assert(failure.getCause eq cause) } + + it should "stop the doubling at maxDelayMillis" in { + val delays = ListBuffer.empty[Long] + intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 5, 200L, noopRetryHook, delays += _, 500L) { + throw new RuntimeException("down") + } + } + assert(delays.toList == List(200L, 400L, 500L, 500L)) + } + + it should "clamp an initial delay that already exceeds maxDelayMillis" in { + val delays = ListBuffer.empty[Long] + intercept[RuntimeException] { + RetryUtil.withBackoff("reach the store", 3, 800L, noopRetryHook, delays += _, 500L) { + throw new RuntimeException("down") + } + } + assert(delays.toList == List(500L, 500L)) + } + + it should "let a failure shouldRetry rejects through unwrapped, spending no further attempts" in { + val delays = ListBuffer.empty[Long] + var attempts = 0 + val cause = new IllegalStateException("no such store") + val failure = intercept[IllegalStateException] { + RetryUtil.withBackoff( + "reach the store", + 5, + 200L, + noopRetryHook, + delays += _, + shouldRetry = _.getMessage == "transient" + ) { + attempts += 1 + throw cause + } + } + assert(failure eq cause) + assert(attempts == 1) + assert(delays.isEmpty) + } + + it should "retry accepted failures yet stop the moment a rejected one appears" in { + // A known transient signal is waited out, but any other failure mid-sequence is a real + // answer and must surface at once rather than be retried alongside it. + val delays = ListBuffer.empty[Long] + var attempts = 0 + val terminal = new IllegalStateException("no such store") + val failure = intercept[IllegalStateException] { + RetryUtil.withBackoff( + "reach the store", + 5, + 200L, + noopRetryHook, + delays += _, + shouldRetry = _.getMessage == "transient" + ) { + attempts += 1 + if (attempts < 3) throw new RuntimeException("transient") + throw terminal + } + } + assert(failure eq terminal) + assert(attempts == 3) + assert(delays.toList == List(200L, 400L)) + } }