kunwp1 commented on code in PR #7744:
URL: https://github.com/apache/texera/pull/7744#discussion_r3810542053
##########
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
Review Comment:
Optional but you can drop these two counters and use `requests.count()`
instead.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -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")
+ }
Review Comment:
There is a existing `RetryUtil.withBackOff` so maybe you can extend this
util function for this code
##########
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
Review Comment:
Might be cleaner if you bundle these three variables to one object with
final case class
##########
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)) {
Review Comment:
Too many redundant `isDelete` condition. Move it into one branch.
##########
amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala:
##########
@@ -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
Review Comment:
Better to swap the if condition to save one wasted parse
--
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]
