mengw15 commented on code in PR #7539:
URL: https://github.com/apache/texera/pull/7539#discussion_r3762242279
##########
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:
CI proved the replacement half of this right: amber's integration
`IcebergDocumentSpec` wrap-and-restores the shared catalog around a spy, and
closing the displaced instance handed every later suite a dead catalog (10
failures). Fixed in 98f88e787: the listener now closes only entries Guava
actually evicted (`wasEvicted()`), never an explicitly replaced one — the
replacing caller owns what it displaced. For the evicted path the earlier
argument stands: a just-resolved entry is the most recently used, so
mid-operation closure stays out of reach 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:
Superseded in 98f88e787: with the listener no longer closing on replacement,
the compare-and-swap pair became unnecessary — `replaceInstance` is a single
atomic `put` again.
--
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]
