yuqi1129 commented on code in PR #11846:
URL: https://github.com/apache/gravitino/pull/11846#discussion_r3570940423
##########
core/src/main/java/org/apache/gravitino/catalog/CapabilityHelpers.java:
##########
@@ -47,17 +47,52 @@
import org.apache.gravitino.rel.partitions.Partition;
import org.apache.gravitino.rel.partitions.Partitions;
import org.apache.gravitino.rel.partitions.RangePartition;
+import org.apache.gravitino.utils.ThrowableFunction;
public class CapabilityHelpers {
- public static Capability getCapability(NameIdentifier ident, CatalogManager
catalogManager) {
+ /**
+ * Executes {@code fn} with the {@link Capability} of the catalog identified
by {@code ident},
+ * inside the catalog's classloader boundary. This prevents the catalog's
{@link
+ * org.apache.gravitino.utils.IsolatedClassLoader} from being closed while
capability methods are
+ * executing.
+ *
+ * <p>If the first attempt encounters a stale (already-closed) {@link
+ * CatalogManager.CatalogWrapper}, the wrapper is evicted from the cache and
the call is retried
+ * once with a freshly loaded wrapper.
+ *
+ * <p>Callers should use this method instead of calling {@code
capabilities()} on a raw wrapper
+ * invoke {@link Capability} methods so that the classloader lifecycle is
properly bounded.
+ *
+ * @param ident any {@link NameIdentifier} that belongs to the target catalog
+ * @param catalogManager the catalog manager used to load the catalog
+ * @param fn function to execute with the catalog's {@link Capability}
+ * @param <R> return type
+ * @return the result of {@code fn}
+ */
+ public static <R> R withCapability(
+ NameIdentifier ident, CatalogManager catalogManager,
ThrowableFunction<Capability, R> fn) {
NameIdentifier catalogIdent = getCatalogIdentifier(ident);
- CatalogManager.CatalogWrapper c =
catalogManager.loadCatalogAndWrap(catalogIdent);
- try {
- return c.capabilities();
- } catch (Exception e) {
- throw new RuntimeException("Failed to get capabilities for catalog: " +
catalogIdent, e);
+ RuntimeException closedException = null;
+ for (int i = 0; i < 2; i++) {
+ CatalogManager.CatalogWrapper c =
catalogManager.loadCatalogAndWrap(catalogIdent);
+ try {
+ return c.doWithCapabilityOps(fn);
+ } catch (IllegalStateException e) {
+ if (c.isClosed() && i == 0) {
Review Comment:
This retry condition cannot distinguish between two very different
situations:
1. `doWithCapabilityOps` threw `IllegalStateException` **at entry** because
the wrapper was already closed (safe to retry — `fn` never ran), and
2. `fn` **itself** threw an `IllegalStateException` mid-execution while the
wrapper happened to be concurrently evicted/closed (`isClosed()` becomes true).
In case 2 the whole `fn` is re-executed. Since several call sites now pass
the entire dispatcher operation as `fn` (e.g.
`TableNormalizeDispatcher.createTable`/`alterTable`,
`PartitionNormalizeDispatcher.addPartition`/`dropPartition`), a non-idempotent
write that partially took effect could be **replayed**, producing duplicate
writes or confusing `AlreadyExists` errors.
Suggestion: have `doWithCapabilityOps` throw a dedicated exception type at
entry (e.g. a package-private `CatalogWrapperClosedException extends
IllegalStateException`) and retry only on that type, so exceptions thrown by
`fn` are never conflated with "wrapper closed at entry". (If `fn` is restricted
to pure normalization as suggested in the other comment, this risk disappears
as well, but the dedicated exception type is still cleaner than the
`isClosed()` heuristic.)
##########
core/src/main/java/org/apache/gravitino/catalog/TableNormalizeDispatcher.java:
##########
@@ -74,26 +74,33 @@ public Table createTable(
SortOrder[] sortOrders,
Index[] indexes)
throws NoSuchSchemaException, TableAlreadyExistsException {
- Capability capability = getCapability(ident, catalogManager);
- return dispatcher.createTable(
- applyCapabilities(ident, Capability.Scope.TABLE, capability),
- applyCapabilities(columns, capability),
- comment,
- properties,
- applyCapabilities(partitions, capability),
- applyCapabilities(distribution, capability),
- applyCapabilities(sortOrders, capability),
- applyCapabilities(indexes, capability));
+ return withCapability(
Review Comment:
Before this change, the capability was only used for name normalization and
`dispatcher.createTable(...)` executed outside the classloader boundary. Now
the **entire dispatcher call chain** runs inside `doWithCapabilityOps`, i.e.
while holding the wrapper's monitor and inside
`classLoader.withClassLoader(...)`. The same applies to `alterTable` below,
`FilesetNormalizeDispatcher.alterFileset`, `ViewNormalizeDispatcher.alterView`,
and all six methods of `PartitionNormalizeDispatcher`.
Consequences:
- All these operations are **serialized per catalog** on one monitor: a
single slow Hive Metastore RPC (e.g. `listPartitions` over tens of thousands of
partitions) blocks every `createTable`/`alterTable`/normalization on the same
catalog.
- `close()` (invoked from Caffeine's removal listener) blocks on the same
monitor for the duration of the in-flight operation; combined with
`CallerRunsPolicy` this can stall the eviction/caller threads.
- Calling external systems while holding the lock enlarges the deadlock
surface (tree locks acquired inside `fn`, cross-catalog nested
`withCapability`).
Suggestion: keep `fn` limited to normalization (which is CPU-only and fast)
and invoke the dispatcher outside the boundary, e.g.:
```java
Pair<NameIdentifier, TableChange[]> normalized =
withCapability(ident, catalogManager,
cap -> Pair.of(applyCaseSensitive(ident, Capability.Scope.TABLE,
cap),
applyCapabilities(cap, changes)));
return dispatcher.alterTable(normalized.getLeft(), normalized.getRight());
```
The `Capability` object still never escapes the boundary, lock hold time
drops back to microseconds, and the retry-replay risk raised in the
`CapabilityHelpers` comment disappears because `fn` becomes a pure function.
--
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]
