You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/20 01:29:16 UTC
(texera) branch gh-readonly-queue/main/pr-7744-68c5774a827f872c99f22c1b64060abee835cde1 created (now c8b9495d96)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7744-68c5774a827f872c99f22c1b64060abee835cde1
in repository https://gitbox.apache.org/repos/asf/texera.git
at c8b9495d96 feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse (#7744)
This branch includes the following new commits:
new c8b9495d96 feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse (#7744)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse (#7744)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7744-68c5774a827f872c99f22c1b64060abee835cde1
in repository https://gitbox.apache.org/repos/asf/texera.git
commit c8b9495d96e90b93873993bce2ba8898b3280941
Author: Meng Wang <me...@uci.edu>
AuthorDate: Thu Aug 20 01:28:54 2026 +0000
feat(amber): wait out Lakekeeper's asynchronous purge when deleting a warehouse (#7744)
### What changes were proposed in this PR?
`LakekeeperClient.deleteWarehouseEmptyFirst` empties the warehouse by
dropping 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 in practice the
first delete of any warehouse that has ever stored execution results
always failed; a retry seconds later succeeded (details in #7742).
- **Retry the final warehouse delete on exactly this conflict**: `409`
with `error.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.
- The bound and delays are constructor parameters with defaults, so
production call sites are unchanged and the spec injects a zero initial
delay — doubling zero stays zero, so the tests contain no real sleeps
and stay deterministic.
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; `deleteWarehouseEmptyFirst` introduced 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 with `409 WarehouseHasUnfinishedTasks` twice 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.
- Full spec run locally: 8/8 passed;
`WorkflowExecutionService/scalafmtCheck` (main + Test) passes.
- The initial retry delay is injected as 0 in all tests — no real
sleeps, deterministic across runs.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-opus-4-8)
---
.../texera/web/service/LakekeeperClient.scala | 68 +++++++++++++++-
.../texera/web/service/LakekeeperClientSpec.scala | 92 ++++++++++++++++++++--
.../org/apache/texera/common/util/RetryUtil.scala | 34 +++++---
.../apache/texera/common/util/RetryUtilSpec.scala | 73 ++++++++++++++++-
4 files changed, 241 insertions(+), 26 deletions(-)
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 7fdafc0df7..ed95d8e4f4 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 ac047c590c..1451720eb0 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, "<html>gateway conflict</html>")
+ 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 3ebf094b4e..63645a737c 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 828cb8d01e..4ac630a005 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))
+ }
}