mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3755525391
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -36,11 +39,34 @@ import scala.collection.mutable
* Only the REST catalog varies by warehouse; the hadoop and postgres
catalogs are warehouse-agnostic
* and ignore the warehouse argument.
*
- * Access is synchronized because the same JVM serves multiple warehouses
concurrently.
+ * The cache is bounded (#7290): per-user warehouses (#6870) make the set of
catalogs a
+ * long-lived JVM touches unbounded, and each REST catalog holds an HTTP
client. Entries
+ * fall out by size or idleness and are closed by the removal listener; the
next access
+ * simply rebuilds one. Callers must therefore resolve their catalog per use
instead of
+ * holding one across an execution (see IcebergDocument / IcebergTableWriter).
*/
-object IcebergCatalogInstance {
+object IcebergCatalogInstance extends LazyLogging {
- private val catalogs = mutable.Map.empty[String, Catalog]
+ // Sizing mirrors HuggingFaceModelResource's bounded-cache precedent:
generous enough
+ // that eviction never hits a warehouse in active use, small enough to bound
the JVM.
+ private val CatalogCacheMaxSize = 64L
+ private val CatalogCacheExpireAfterAccessMinutes = 60L
+
+ private val catalogs: Cache[String, Catalog] = CacheBuilder
+ .newBuilder()
+ .maximumSize(CatalogCacheMaxSize)
+ .expireAfterAccess(CatalogCacheExpireAfterAccessMinutes, TimeUnit.MINUTES)
+ .removalListener(new RemovalListener[String, Catalog] {
+ override def onRemoval(notification: RemovalNotification[String,
Catalog]): Unit =
+ notification.getValue match {
+ case closeable: AutoCloseable =>
+ Try(closeable.close()).failed.foreach(error =>
+ logger.warn(s"failed to close evicted catalog
'${notification.getKey}'", error)
Review Comment:
The long-lived pin this describes was real, but it was the writer's eager
`table` val — fixed in f0e5da334 to resolve per flush, which also keeps the
entry's access time fresh for the whole execution. What remains is the
within-one-operation window: a just-resolved entry is the cache's most recently
used, so size eviction cannot pick it, and idle expiry needs 60 idle minutes.
Closing it mid-operation would take ~64 new warehouses materializing inside a
single operation; a lease/refcount layer seems heavier than that risk warrants
at this scale.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -102,8 +133,14 @@ object IcebergCatalogInstance {
* @param catalog the catalog to cache.
* @param warehouse the warehouse to cache it under; `None` uses the
configured default.
*/
- def replaceInstance(catalog: Catalog, warehouse: Option[String] = None):
Unit =
- synchronized {
- catalogs(cacheKey(warehouse.getOrElse(defaultWarehouse))) = catalog
+ def replaceInstance(catalog: Catalog, warehouse: Option[String] = None):
Unit = {
+ val key = cacheKey(warehouse.getOrElse(defaultWarehouse))
+ // Guava reports a same-value put as a replacement, which would fire the
removal
+ // listener and close a catalog that is still installed: the shared test
catalog
+ // is ensure()d repeatedly (and under several names) by parallel suites.
Skip the
+ // no-op re-put so only a genuine replacement closes the previous catalog.
+ if (catalogs.getIfPresent(key) ne catalog) {
+ catalogs.put(key, catalog)
}
Review Comment:
Done in f0e5da334: `replaceInstance` now goes through `asMap`'s
`putIfAbsent` + `replace(key, previous, catalog)`, so a lost race leaves the
winner installed and only a genuine replacement fires the listener.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/result/iceberg/IcebergTableWriter.scala:
##########
@@ -50,13 +52,17 @@ import scala.collection.mutable.ArrayBuffer
*/
private[storage] class IcebergTableWriter[T](
val writerIdentifier: String,
- val catalog: Catalog,
+ val warehouse: Option[String],
val tableNamespace: String,
val tableName: String,
val tableSchema: Schema,
val serde: (org.apache.iceberg.Schema, T) => Record
) extends BufferedItemWriter[T] {
+ // Resolved per use (#7290): the catalog cache is bounded and closes evicted
entries,
+ // so the writer must not pin one across its lifetime.
+ private def catalog: Catalog = IcebergCatalogInstance.getInstance(warehouse)
Review Comment:
Good catch — fixed in f0e5da334. The eager val pinned a `Table` across the
writer's lifetime, exactly the reference the cache could close. `flushBuffer`
now resolves the table per flush; commits already pay a REST round trip, and
the per-flush lookup keeps the warehouse's entry live during long executions.
##########
common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstanceSpec.scala:
##########
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.core.storage
+
+import org.apache.texera.amber.core.storage.result.iceberg.IcebergDocument
+import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple}
+import org.apache.texera.amber.util.IcebergUtil
+import org.apache.iceberg.Table
+import org.apache.iceberg.catalog.{Catalog, Namespace, TableIdentifier}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/**
+ * Spec for the bounded catalog cache (#7290): a genuine replacement closes
the
+ * catalog it displaces, a same-instance re-registration does not (that is
what
+ * [[LocalHadoopIcebergCatalog.ensure]] relies on), and holders resolve their
+ * catalog per use so a replacement is visible immediately.
+ *
+ * Size-based eviction is deliberately not exercised: forcing it would flood
the
+ * JVM-wide cache that parallel suites share and could evict their live
catalog.
+ * The close-on-removal wiring it would exercise is pinned by the replacement
+ * cases below, which Guava routes through the same removal listener.
Review Comment:
Done in f0e5da334: cache construction is behind a package-private
`buildCatalogCache(maximumSize, expireAfterAccess, ticker)`; the spec pins size
eviction and idle expiry on isolated instances with a manual ticker, without
touching the JVM-wide cache.
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/IcebergCatalogInstance.scala:
##########
@@ -70,9 +96,14 @@ object IcebergCatalogInstance {
*/
def getInstance(warehouse: Option[String] = None): Catalog = {
val name = warehouse.getOrElse(defaultWarehouse)
- synchronized {
- catalogs.getOrElseUpdate(cacheKey(name), createCatalog(name))
- }
+ // get(key, loader) locks per key, not globally: a cache miss's REST config
+ // round trip no longer blocks lookups of other warehouses.
+ catalogs.get(
+ cacheKey(name),
+ new Callable[Catalog] {
+ override def call(): Catalog = createCatalog(name)
+ }
+ )
Review Comment:
Done in f0e5da334: `getInstance` unwraps `UncheckedExecutionException` /
`ExecutionException` / `ExecutionError` back to the loader's original
exception, pinned by spec cases for both runtime and checked failures.
--
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]
