Copilot commented on code in PR #7539: URL: https://github.com/apache/texera/pull/7539#discussion_r3755483996
########## 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: The PR's core size and idle-expiration behavior remains untested; replacement only exercises the listener, so removing `maximumSize`, changing the limit/TTL, or breaking time-based eviction would still pass. Extract cache construction behind a package-private factory with configurable size/ticker, then test size and expiration on an isolated cache rather than mutating the JVM-wide singleton. ########## 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: Closing a removed entry immediately is unsafe because `Cache.get` returns an unpinned reference. A caller can obtain catalog A, then another warehouse insertion can evict and close A before `tableExists`/`loadTable` runs; loaded `Table` objects and iterators can also continue using A's REST client after eviction. Resolving on each access does not close this lifetime gap. Use a borrow/lease (or equivalent quiescent-retirement mechanism) and close an evicted catalog only after no in-flight catalog or derived table operation still uses it. ########## 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: This accessor is not actually resolved for each writer use: it is evaluated once by the eager `private val table` at line 76. That `Table` retains REST table operations backed by the catalog client, so eviction or replacement after writer construction can close the client before `flushBuffer` commits. Resolve the table for each flush, or retain a cache lease until the writer closes. ########## 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: `Cache.get` wraps loader failures (`UncheckedExecutionException` for runtime exceptions, `ExecutionException` for checked exceptions, and `ExecutionError` for errors). Previously `createCatalog` failures such as `RESTException` or the unsupported-type `IllegalArgumentException` propagated directly, so this silently changes `getInstance`'s error contract and can break existing error handling. Unwrap these Guava wrappers before returning. ########## 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: This check-then-put is not atomic. If two threads re-register the same catalog while another value is installed, both can observe the old value; the first installs the shared catalog and the second `put` reports that same shared instance as replaced, causing the listener to close the value that remains cached. Use an atomic compare-and-replace loop and add a concurrent regression test. -- 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]
