mengw15 commented on code in PR #7744:
URL: https://github.com/apache/texera/pull/7744#discussion_r3816571544
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -20,15 +20,37 @@
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
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 `retries` retries wait
0.2+0.4+0.8+1.6+3.2+5+5s
+ * ≈ 16s in total. Overridable for tests (a 0 initial delay keeps the spec
free of
+ * real sleeps — doubling 0 stays 0).
+ */
+ final case class PurgeWaitPolicy(
+ retries: Int = 7,
Review Comment:
Done — `maxAttempts = 8`, the `+1` is gone.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -126,12 +153,43 @@ 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 = "delete warehouse",
+ maxAttempts = purgeWait.retries + 1,
+ initialDelayMillis = purgeWait.initialDelayMillis,
+ onRetry = attempt => logger.info(attempt.message),
Review Comment:
Done.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -126,12 +153,43 @@ 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 = "delete warehouse",
+ maxAttempts = purgeWait.retries + 1,
+ initialDelayMillis = purgeWait.initialDelayMillis,
+ onRetry = attempt => logger.info(attempt.message),
+ maxDelayMillis = purgeWait.maxDelayMillis,
+ shouldRetry = _.isInstanceOf[UnfinishedTasksConflictException]
+ ) {
+ val response =
Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString()
+ if (response.getStatus != 404) {
+ if (isUnfinishedTasksConflict(response.getStatus, response.getBody)) {
+ throw new UnfinishedTasksConflictException(response.getStatus,
response.getBody)
+ }
+ failOn(response.getStatus, response.getBody, "delete warehouse")
+ }
Review Comment:
Done — one flat match; the check is body-only now.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -126,12 +153,43 @@ 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 = "delete warehouse",
+ maxAttempts = purgeWait.retries + 1,
+ initialDelayMillis = purgeWait.initialDelayMillis,
+ onRetry = attempt => logger.info(attempt.message),
+ maxDelayMillis = purgeWait.maxDelayMillis,
+ shouldRetry = _.isInstanceOf[UnfinishedTasksConflictException]
+ ) {
+ val response =
Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString()
+ if (response.getStatus != 404) {
+ if (isUnfinishedTasksConflict(response.getStatus, response.getBody)) {
+ throw new UnfinishedTasksConflictException(response.getStatus,
response.getBody)
+ }
+ failOn(response.getStatus, response.getBody, "delete warehouse")
+ }
}
}
+ /** Tags the one retryable delete failure so the backoff predicate can
single it out. */
+ private class UnfinishedTasksConflictException(status: Int, body: String)
Review Comment:
Done — the exception takes only the body.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -126,12 +153,43 @@ 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 = "delete warehouse",
+ maxAttempts = purgeWait.retries + 1,
+ initialDelayMillis = purgeWait.initialDelayMillis,
+ onRetry = attempt => logger.info(attempt.message),
+ maxDelayMillis = purgeWait.maxDelayMillis,
+ shouldRetry = _.isInstanceOf[UnfinishedTasksConflictException]
+ ) {
+ val response =
Unirest.delete(s"$managementBase/warehouse/$warehouseId").asString()
+ if (response.getStatus != 404) {
+ if (isUnfinishedTasksConflict(response.getStatus, response.getBody)) {
+ throw new UnfinishedTasksConflictException(response.getStatus,
response.getBody)
+ }
+ failOn(response.getStatus, response.getBody, "delete warehouse")
+ }
}
}
+ /** Tags the one retryable delete failure so the backoff predicate can
single it out. */
+ private class UnfinishedTasksConflictException(status: Int, body: String)
+ extends RuntimeException(s"Lakekeeper delete warehouse failed (HTTP
$status): $body")
Review Comment:
Done — extracted `DeleteWarehouseAction`.
##########
amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala:
##########
@@ -121,6 +158,13 @@ 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";,
Review Comment:
Done — extracted `stubCatalogUri`.
##########
amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala:
##########
@@ -168,4 +212,40 @@ 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")
+ // 1 initial attempt + 3 retries, then fail -- the wait is bounded.
Review Comment:
Dropped.
##########
amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala:
##########
@@ -168,4 +212,40 @@ 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")
+ // 1 initial attempt + 3 retries, then fail -- the wait is bounded.
+ deleteAttempts(alwaysBusyWarehouseId) 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")
+ deleteAttempts(malformedConflictWarehouseId) 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")
+ deleteAttempts(otherConflictWarehouseId) shouldBe 1
+ }
Review Comment:
Done — shared `assertFailsWithoutRetry`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
