kunwp1 commented on code in PR #7744:
URL: https://github.com/apache/texera/pull/7744#discussion_r3814462024
##########
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:
Can you refactor this code by flattening to a three-outcome match? It's too
hard to read because it has two nesting levels. Then
`isUnfinishedTasksConflict` can be much simpler.
##########
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:
I would prefer to rename it to maxAttempts = 8 and drop the +1 in the code
below when you use it.
##########
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:
Might not need this line after the change above.
##########
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:
Refactor this code because there are duplicate codes.
##########
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:
Refactor this line and line 158
##########
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:
Drop status because this is only reachable by 409.
##########
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:
I think "delete warehouse" string appears so many time. You can refactor
this.
##########
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:
Use WARN for retries.
##########
amber/src/test/scala/org/apache/texera/web/service/LakekeeperClientSpec.scala:
##########
@@ -65,13 +71,44 @@ 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 if (exchange.getRequestMethod == "DELETE") {
+ val path = exchange.getRequestURI.getPath
+ if (path.endsWith(racingWarehouseId.toString)) {
+ if (deleteAttempts(racingWarehouseId) <= 2) respond(exchange, 409,
unfinishedTasksBody)
+ else respond(exchange, 204, "")
+ } else if (path.endsWith(alwaysBusyWarehouseId.toString)) {
+ respond(exchange, 409, unfinishedTasksBody)
+ } else if (path.endsWith(malformedConflictWarehouseId.toString)) {
+ // A 409 whose body isn't the JSON envelope the type check reads.
+ respond(exchange, 409, "<html>gateway conflict</html>")
+ } else if (path.endsWith(otherConflictWarehouseId.toString)) {
+ respond(
+ exchange,
+ 409,
+ """{"error":{"message":"warehouse is in
use","type":"Conflict","code":409}}"""
+ )
+ } else {
+ respond(exchange, 200, "{}")
+ }
} else {
respond(exchange, 200, "{}")
}
Review Comment:
Too many branch depth. I think you can make it cleaner.
--
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]
