LuciferYang commented on code in PR #14741:
URL: https://github.com/apache/gravitino/pull/14741#discussion_r4495037120
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/GravitinoCatalogManager.java:
##########
@@ -152,29 +166,59 @@ public void close() {
}
public Catalog getGravitinoCatalogInfo(String name) {
- GravitinoIdentity identity = currentIdentity();
try {
- return gravitinoCatalogs.get(cacheKey(identity, name), key ->
loadCatalog(identity, name));
+ return getGravitinoCatalogInfo(currentIdentity(), name);
} catch (Exception e) {
LOG.error(String.format("Load catalog %s failed", name), e);
throw new RuntimeException(e);
}
}
+ /**
+ * Returns the catalog for the given identity, loaded through that
identity's current client.
+ *
+ * <p>Spark catalogs call this before every Gravitino request instead of
keeping the returned
+ * catalog: the lookup refreshes the client's cache entry, so a client in
use is not evicted for
+ * being idle, and a catalog loaded through a client that has since been
evicted and closed is
+ * loaded again through the current one.
+ *
+ * <p>Gravitino exceptions are thrown as is, so callers can still tell them
apart.
+ *
+ * @param identity the identity the catalog is loaded for
+ * @param name the catalog name
+ * @return the catalog, bound to the identity's current client
+ */
+ Catalog getGravitinoCatalogInfo(GravitinoIdentity identity, String name) {
+ // Spark catalogs keep a reference to this manager, so fail instead of
building a new client.
+ Preconditions.checkState(!isClosed, "GravitinoCatalogManager is already
closed");
+ GravitinoClient client = getClient(identity);
+ String key = cacheKey(identity, name);
+ CachedCatalog cached =
+ gravitinoCatalogs.get(
+ key, k -> new CachedCatalog(client, loadCatalog(client, identity,
name)));
+ if (cached.client != client) {
+ cached = new CachedCatalog(client, loadCatalog(client, identity, name));
+ gravitinoCatalogs.put(key, cached);
Review Comment:
Done in da55b517d: the stale-entry reload now runs in
`gravitinoCatalogs.asMap().compute(...)`, which reloads only when the entry is
still not bound to the current client, so a caller that waited for another
caller's reload reuses it. `testConcurrentLookupsReloadOnce` holds the first
reload with a latch until the second caller reaches the lookup and checks that
there is one `loadCatalog` and both callers get the same catalog; with the
previous `put` it saw two.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/catalog/BaseCatalog.java:
##########
@@ -192,12 +193,24 @@ public String name() {
return catalogName;
}
+ /**
+ * Returns the Gravitino catalog that serves schema, table and function
requests. Look it up for
+ * each request instead of keeping it: the lookup resets the idle timer of
the Gravitino client in
+ * the manager's cache, and returns a catalog bound to a live client after
the cache has evicted
+ * and closed an earlier one.
Review Comment:
Done in da55b517d and e1d7bb823: the Javadoc of `gravitinoCatalog()` says
the lookup does not keep the client alive for the rest of the request, so a
request still running when its client is evicted, by size or by idle time, can
fail and the next lookup gets a new client. The manager's Javadoc no longer
says a client in use is never evicted, and the PR description says the same.
--
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]
