feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse - #7744
Open
mengw15 wants to merge 6 commits into
Open
feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse#7744mengw15 wants to merge 6 commits into
mengw15 wants to merge 6 commits into
Conversation
Contributor
Backport auto-label reportThis
|
Contributor
Automated Reviewer SuggestionsBased on the
|
… warehouse deleteWarehouseEmptyFirst drops every table with purgeRequested=true, then immediately deletes the warehouse entity. Lakekeeper purges the dropped tables' data files asynchronously (task queue `tabular_purge`) and 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, so the first delete of any warehouse that had ever stored execution results always failed; a retry seconds later succeeded. Retry the final warehouse delete on exactly that conflict -- 409 with error.type WarehouseHasUnfinishedTasks -- with a bounded pause (default 10 retries x 2s; the queue normally drains within seconds). Every other error, including any other 409, still fails immediately, and 404 stays the idempotent goal state. The bound and delay are constructor parameters with defaults, so production call sites are unchanged and the spec injects a zero delay -- no real sleeps in the tests. LakekeeperClientSpec covers the three outcomes against its in-process stub: 409-409-204 succeeds with exactly 3 attempts (failed before the fix on the first 409), a never-draining queue fails after the bounded 1+3 attempts, and a 409 of any other type fails on the first attempt with no retry. Found while testing the flag-gated per-user warehouse feature; no deployment is affected because the flag defaults to off. Closes apache#7742.
mengw15
changed the title
mengw15
force-pushed
the
fix/7742-warehouse-delete-purge-retry
branch
from
August 18, 2026 00:31
e82ece1 to
6ce6cba
Compare
Contributor
|
| config | throughput | MB/s | latency | max Δ latest / 7d | |
|---|---|---|---|---|---|
| 🔴 | bs=10 sw=10 sl=64 | 566 | 0.346 | 17,223/22,597/22,597 us | 🔴 -6.9% / 🔴 +44.9% |
| ⚪ | bs=100 sw=10 sl=64 | 1,229 | 0.75 | 80,101/93,758/93,758 us | ⚪ within ±5% / 🟢 +23.3% |
| ⚪ | bs=1000 sw=10 sl=64 | 1,394 | 0.851 | 721,079/750,023/750,023 us | ⚪ within ±5% / 🟢 +34.4% |
Baseline details
Latest main 60300e3 from same runner
| config | metric | PR | latest main | 7d avg | Δ latest | Δ 7d |
|---|---|---|---|---|---|---|
| bs=10 sw=10 sl=64 | throughput | 566 tuples/sec | 608 tuples/sec | 771.02 tuples/sec | -6.9% | -26.6% |
| bs=10 sw=10 sl=64 | MB/s | 0.346 MB/s | 0.371 MB/s | 0.471 MB/s | -6.7% | -26.5% |
| bs=10 sw=10 sl=64 | p50 | 17,223 us | 16,677 us | 12,749 us | +3.3% | +35.1% |
| bs=10 sw=10 sl=64 | p95 | 22,597 us | 24,171 us | 15,594 us | -6.5% | +44.9% |
| bs=10 sw=10 sl=64 | p99 | 22,597 us | 24,171 us | 19,320 us | -6.5% | +17.0% |
| bs=100 sw=10 sl=64 | throughput | 1,229 tuples/sec | 1,260 tuples/sec | 997.05 tuples/sec | -2.5% | +23.3% |
| bs=100 sw=10 sl=64 | MB/s | 0.75 MB/s | 0.769 MB/s | 0.609 MB/s | -2.5% | +23.2% |
| bs=100 sw=10 sl=64 | p50 | 80,101 us | 79,329 us | 100,339 us | +1.0% | -20.2% |
| bs=100 sw=10 sl=64 | p95 | 93,758 us | 94,594 us | 106,781 us | -0.9% | -12.2% |
| bs=100 sw=10 sl=64 | p99 | 93,758 us | 94,594 us | 113,101 us | -0.9% | -17.1% |
| bs=1000 sw=10 sl=64 | throughput | 1,394 tuples/sec | 1,382 tuples/sec | 1,037 tuples/sec | +0.9% | +34.4% |
| bs=1000 sw=10 sl=64 | MB/s | 0.851 MB/s | 0.843 MB/s | 0.633 MB/s | +0.9% | +34.4% |
| bs=1000 sw=10 sl=64 | p50 | 721,079 us | 726,110 us | 971,388 us | -0.7% | -25.8% |
| bs=1000 sw=10 sl=64 | p95 | 750,023 us | 773,648 us | 1,013,249 us | -3.1% | -26.0% |
| bs=1000 sw=10 sl=64 | p99 | 750,023 us | 773,648 us | 1,039,879 us | -3.1% | -27.9% |
Raw CSV
config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,353.31,200,128000,566,0.346,17223.34,22596.74,22596.74
1,100,10,64,20,1626.79,2000,1280000,1229,0.750,80101.19,93758.24,93758.24
2,1000,10,64,20,14344.90,20000,12800000,1394,0.851,721079.44,750022.92,750022.92
Contributor
There was a problem hiding this comment.
Pull request overview
Adds bounded retry handling for Lakekeeper’s asynchronous warehouse purge conflicts.
Changes:
- Retries only
409 WarehouseHasUnfinishedTasks. - Adds deterministic success, exhaustion, and unrelated-conflict tests.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
LakekeeperClient.scala |
Implements bounded warehouse-delete retries. |
LakekeeperClientSpec.scala |
Tests retry and failure behavior. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7744 +/- ##
============================================
+ Coverage 91.02% 91.22% +0.19%
- Complexity 4454 4472 +18
============================================
Files 1174 1171 -3
Lines 47146 47135 -11
Branches 5287 5287
============================================
+ Hits 42916 42998 +82
+ Misses 2550 2457 -93
Partials 1680 1680
*This pull request uses carry forward flags. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
The retry only fires for a 409 whose body carries the WarehouseHasUnfinishedTasks type, and the type check parses that body -- so a 409 with a non-JSON body (a gateway error page, say) must read as 'not the purge conflict' and fail immediately rather than be waited out as if it were transient. Assert it fails on the first attempt.
…ueue The wait was a fixed 2s x 10. That made the common case -- a purge that drains almost immediately -- cost the caller a full 2s before the delete was retried, on a request a user is waiting on, while still issuing 11 requests when the queue is genuinely slow. Double the pause instead, from 200ms up to a 5s cap over 7 retries: 0.2+0.4+0.8+1.6+3.2+5+5s, so a fast purge returns in ~200ms, a slow one issues fewer requests, and the overall bound drops from 20s to ~16s. Tests still inject a zero initial delay -- doubling zero stays zero, so they contain no real sleeps.
kunwp1
reviewed
Aug 19, 2026
- Bundle the three retry knobs into `LakekeeperClient.PurgeWaitPolicy`, so the constructor keeps two parameters and the defaults live in one place. - Check the retry budget before parsing the response body: on the attempt that exhausts it, the parse was thrown away. - Count the spec's delete attempts off the request log the stub already keeps, dropping the three counters and their per-test reset. - Dispatch the stub's DELETE cases under a single method check.
Review follow-up: rather than a hand-written retry loop, extend the shared util with what the purge wait needs -- an optional retry predicate (shouldRetry) and a per-wait delay cap (maxDelayMillis), both defaulted so the existing call sites keep their behavior -- and have LakekeeperClient tag the one retryable failure (409 WarehouseHasUnfinishedTasks) with a private exception for the predicate to match.
kunwp1
reviewed
Aug 19, 2026
kunwp1
left a comment
kunwp1
left a comment
Contributor
There was a problem hiding this comment.
The main functionality LGTM! Left some more comments to cleanup the codes.
| * real sleeps — doubling 0 stays 0). | ||
| */ | ||
| final case class PurgeWaitPolicy( | ||
| retries: Int = 7, |
Contributor
There was a problem hiding this comment.
I would prefer to rename it to maxAttempts = 8 and drop the +1 in the code below when you use it.
Contributor
Author
There was a problem hiding this comment.
Done — maxAttempts = 8, the +1 is gone.
| description = "delete warehouse", | ||
| maxAttempts = purgeWait.retries + 1, | ||
| initialDelayMillis = purgeWait.initialDelayMillis, | ||
| onRetry = attempt => logger.info(attempt.message), |
Comment on lines
+172
to
+177
| if (response.getStatus != 404) { | ||
| if (isUnfinishedTasksConflict(response.getStatus, response.getBody)) { | ||
| throw new UnfinishedTasksConflictException(response.getStatus, response.getBody) | ||
| } | ||
| failOn(response.getStatus, response.getBody, "delete warehouse") | ||
| } |
Contributor
There was a problem hiding this 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.
Contributor
Author
There was a problem hiding this comment.
Done — one flat match; the check is body-only now.
| } | ||
|
|
||
| /** Tags the one retryable delete failure so the backoff predicate can single it out. */ | ||
| private class UnfinishedTasksConflictException(status: Int, body: String) |
Contributor
There was a problem hiding this comment.
Drop status because this is only reachable by 409.
Contributor
Author
There was a problem hiding this comment.
Done — the exception takes only the body.
|
|
||
| /** 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") |
Contributor
There was a problem hiding this comment.
I think "delete warehouse" string appears so many time. You can refactor this.
Contributor
Author
There was a problem hiding this comment.
Done — extracted DeleteWarehouseAction.
| // 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", |
Contributor
There was a problem hiding this comment.
Refactor this line and line 158
Contributor
Author
There was a problem hiding this comment.
Done — extracted stubCatalogUri.
| } | ||
| error.getMessage should include("409") | ||
| error.getMessage should include("WarehouseHasUnfinishedTasks") | ||
| // 1 initial attempt + 3 retries, then fail -- the wait is bounded. |
Contributor
There was a problem hiding this comment.
Might not need this line after the change above.
Comment on lines
+234
to
+250
| 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 | ||
| } |
Contributor
There was a problem hiding this comment.
Refactor this code because there are duplicate codes.
Contributor
Author
There was a problem hiding this comment.
Done — shared assertFailsWithoutRetry.
Comment on lines
93
to
114
| } 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, "{}") | ||
| } |
Contributor
There was a problem hiding this comment.
Too many branch depth. I think you can make it cleaner.
Contributor
Author
There was a problem hiding this comment.
Done — flattened to a single match on (method, path).
- PurgeWaitPolicy counts total maxAttempts (8) instead of retries, dropping the +1 at the call site. - The delete outcome is one flat three-case match (gone / retryable purge conflict / fail), so the conflict check reads only the body and the marker exception no longer carries a status. - Retries log at WARN, matching the other withBackoff callers. - "delete warehouse" lives in one constant. - Spec: the stub dispatch is one (method, path) match, the stub URI and the no-retry-409 assertions are shared, and the counted-attempts comment goes.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this PR?
LakekeeperClient.deleteWarehouseEmptyFirstempties the warehouse by dropping every table withpurgeRequested=true, then immediately deletes the warehouse entity. Lakekeeper purges the dropped tables' data files asynchronously (task queuetabular_purge) and 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 answers409 WarehouseHasUnfinishedTasksuntil the queue drains, so in practice the first delete of any warehouse that has ever stored execution results always failed; a retry seconds later succeeded (details in #7742).409witherror.type == "WarehouseHasUnfinishedTasks"is treated as transient and retried with exponential backoff — 200ms doubling up to a 5s cap over 7 retries (0.2+0.4+0.8+1.6+3.2+5+5s ≈ 16s bound). A purge that drains almost immediately, the common case, costs the caller ~200ms rather than a full fixed interval; a genuinely slow one issues fewer requests than a fixed pause would. Every other error — including any other 409, and a 409 whose body isn't the expected JSON envelope — still fails immediately, and 404 stays the idempotent goal state.Found while testing the flag-gated per-user warehouse feature (#6870); no deployment is affected because the flag defaults to off.
Any related issues, documentation, discussions?
Closes #7742. Part of #6870;
deleteWarehouseEmptyFirstintroduced in #7473. Verified against the local Lakekeeper (0.11.0) that its management API exposes no task-query endpoint or force-delete option, so waiting out the documented conflict type is the only client-side path.How was this PR tested?
LakekeeperClientSpec(in-process HTTP stub, no external infra) gains four cases mirroring the issue's repro: the stub answers the warehouse delete with409 WarehouseHasUnfinishedTaskstwice and then 204 — verified failing before the fix (the first 409 threw) and passing after, with the stub asserting exactly 3 delete attempts; a warehouse whose queue never drains fails after the bounded 1+3 attempts; a 409 of any other type fails on the first attempt with no retry; and a 409 whose body is not JSON (a gateway error page) also fails on the first attempt rather than being waited out.WorkflowExecutionService/scalafmtCheck(main + Test) passes.Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-4-8)