jerryshao commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3841065853
##########
server/src/main/java/org/apache/gravitino/server/web/rest/ExceptionHandlers.java:
##########
@@ -212,6 +199,38 @@ public static Response handlePartitionStatsException(
return PartitionStatsExceptionHandler.INSTANCE.handle(type, name, parent,
e);
}
+ private static Response handleTestConnectionException(Exception e, boolean
includeStack) {
+ Throwable throwable = includeStack ? e : null;
+ ErrorResponse response;
+ if (e instanceof IllegalArgumentException) {
+ response = ErrorResponse.illegalArguments(e.getMessage(), throwable);
+
+ } else if (e instanceof ConnectionFailedException) {
+ response = ErrorResponse.connectionFailed(e.getMessage(), throwable);
+
+ } else if (e instanceof UnsupportedOperationException) {
+ response = ErrorResponse.unsupportedOperation(e.getMessage(), throwable);
+
+ } else if (e instanceof NotFoundException) {
+ response = ErrorResponse.notFound(e.getClass().getSimpleName(),
e.getMessage(), throwable);
+
+ } else if (e instanceof AlreadyExistsException) {
+ response =
+ ErrorResponse.alreadyExists(e.getClass().getSimpleName(),
e.getMessage(), throwable);
+
+ } else if (e instanceof NotInUseException) {
+ response = ErrorResponse.notInUse(e.getClass().getSimpleName(),
e.getMessage(), throwable);
+
+ } else {
+ return Utils.internalError(e.getMessage(), e);
+ }
+
Review Comment:
**[Confirmed bug]** The fallback `else` branch still calls
`Utils.internalError(e.getMessage(), e)` with the raw exception `e` instead of
`throwable`, while the six branches above it correctly null the stack via
`throwable = includeStack ? e : null`.
For the sanitized endpoint (`includeStack=false`), any exception outside the
six explicitly handled types — e.g. the generic `RuntimeException` wrapper from
`CatalogManager.testConnection(NameIdentifier)`'s catch-all — falls into this
branch and returns the full stack trace via
`ErrorResponse.internalError(message, e)`, defeating the endpoint's documented
"sanitized failure message" guarantee.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -946,24 +958,43 @@ public boolean dropSchema(NameIdentifier ident, boolean
cascade) throws NonEmpty
}
}
- /**
- * Since the Fileset catalog was completely managed by Gravitino, we don't
need to test the
- * connection
- *
- * @param catalogIdent the name of the catalog.
- * @param type the type of the catalog.
- * @param provider the provider of the catalog.
- * @param comment the comment of the catalog.
- * @param properties the properties of the catalog.
- */
@Override
public void testConnection(
NameIdentifier catalogIdent,
Catalog.Type type,
String provider,
String comment,
Map<String, String> properties) {
- // Do nothing
+ testConnection(catalogIdent);
+ }
+
+ @Override
+ public void testConnection(NameIdentifier catalogIdent) {
+ if (disableFSOps) {
+ throw new UnsupportedOperationException(
+ "Fileset connection testing requires filesystem operations to be
enabled");
+ }
+ if (catalogStorageLocations.isEmpty()) {
+ throw new IllegalArgumentException("Fileset catalog has no catalog-level
location to test");
+ }
+
Review Comment:
**[Plausible]** Inside `testConnection(NameIdentifier)`'s
`probeLocations.forEach`, an `UnsupportedOperationException` from one location
is caught and immediately rethrown, aborting the whole `forEach` and skipping
remaining locations — unlike other exception types, which are aggregated into
the `failures` list.
A catalog with two locations where the first trips an
`UnsupportedOperationException` (e.g. two credential providers both match its
scheme) causes `testConnection()` to abort immediately; the second location's
connectivity is never probed or reported, contradicting the aggregate-failure
contract demonstrated by `testTestConnectionAggregatesLocationFailures` for
ordinary probe failures.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1443,226 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ private Map<String, Path> resolveProbeLocations(NameIdentifier catalogIdent)
{
+ Map<String, Path> resolved = new HashMap<>();
+ catalogStorageLocations.forEach(
+ (locationName, location) ->
+ resolved.put(locationName,
resolveProbeLocation(location.toString(), catalogIdent)));
+ return resolved;
+ }
+
+ private Path resolveProbeLocation(String location, NameIdentifier
catalogIdent) {
+ String resolved = location.replace("{{catalog}}", catalogIdent.name());
+ Matcher matcher = LOCATION_PLACEHOLDER_PATTERN.matcher(resolved);
+ if (matcher.find()) {
+ String staticPrefix = resolved.substring(0, matcher.start());
+ int lastSeparator = staticPrefix.lastIndexOf(SLASH);
+ if (lastSeparator < 0) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no concrete parent that can be
tested");
+ }
+ resolved = staticPrefix.substring(0, lastSeparator + 1);
+ }
+
+ try {
+ URI uri = URI.create(resolved);
+ if (StringUtils.isBlank(resolved)
+ || containsPlaceholder(resolved)
+ || (uri.getScheme() != null
+ && resolved.contains("://")
+ && !"file".equalsIgnoreCase(uri.getScheme())
+ && StringUtils.isBlank(uri.getAuthority()))) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no valid concrete target that can be
tested");
+ }
+ return new Path(resolved);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no valid concrete target that can be
tested", e);
+ }
+ }
+
+ private void probeLocation(Path path) throws Exception {
+ Map<String, String> probeConf = new HashMap<>(conf);
+ probeConf.putAll(
+ FilesetUtil.getUserDefinedFileSystemConfigs(
+ path.toUri(),
+ probeConf,
+ FilesetCatalogPropertiesMetadata.FS_GRAVITINO_PATH_CONFIG_PREFIX));
+
+ String scheme =
+ path.toUri().getScheme() != null
+ ? path.toUri().getScheme()
+ : defaultFileSystemProvider.scheme();
+ FileSystemProvider fileSystemProvider = fileSystemProvidersMap.get(scheme);
+ if (fileSystemProvider == null) {
+ throw new UnsupportedOperationException(
+ String.format("Fileset connection testing does not support scheme
%s", scheme));
+ }
+
+ String credentialHandle = null;
+ Future<Void> probeFuture = null;
+ try {
+ credentialHandle = addVendedCredential(path, scheme, fileSystemProvider,
probeConf);
+ int timeoutSeconds =
+ (int)
+ propertiesMetadata
+ .catalogPropertiesMetadata()
+ .getOrDefault(
+ probeConf,
+
FilesetCatalogPropertiesMetadata.FILESYSTEM_CONNECTION_TIMEOUT_SECONDS);
+ probeFuture =
+ fileSystemExecutor.submit(
+ () -> {
+ try (FileSystem fileSystem =
+ createFileSystem(path, probeConf, scheme,
fileSystemProvider)) {
+ Path qualifiedPath =
+ path.makeQualified(fileSystem.getUri(),
fileSystem.getWorkingDirectory());
+ RemoteIterator<FileStatus> iterator =
+ fileSystem.listStatusIterator(qualifiedPath);
+ if (iterator.hasNext()) {
+ FileStatus first = iterator.next();
+ if (first.isFile() &&
first.getPath().equals(qualifiedPath)) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location must be a directory");
+ }
+ }
+ return null;
+ }
+ });
+ probeFuture.get(timeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ if (probeFuture != null) {
+ probeFuture.cancel(true);
+ }
+ throw new IOException("Filesystem connection probe timed out", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("Filesystem connection probe was interrupted", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Exception) {
+ throw (Exception) cause;
+ }
+ throw new IOException("Filesystem connection probe failed", cause);
+ } finally {
+ InMemoryFileSystemCredentialsProvider.unregister(credentialHandle);
+ }
+ }
+
+ @Nullable
+ private String addVendedCredential(
+ Path path,
+ String scheme,
+ FileSystemProvider fileSystemProvider,
+ Map<String, String> probeConf) {
+ Set<String> credentialTypes =
CredentialUtils.getCredentialProvidersByOrder(() -> conf);
+ if (credentialTypes.isEmpty()) {
+ return null;
+ }
+
+ CatalogCredentialManager manager = credentialManager();
+ List<String> matchingTypes =
+ credentialTypes.stream()
+ .filter(
+ type ->
+ manager
+ .getCredentialProvider(type)
+ .map(provider -> provider.supportsScheme(scheme))
+ .orElse(false))
+ .collect(Collectors.toList());
+ if (matchingTypes.isEmpty()) {
+ return null;
+ }
+ if (matchingTypes.size() > 1) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Multiple credential providers match Fileset location scheme %s:
%s",
+ scheme, matchingTypes));
+ }
+ if (!(fileSystemProvider instanceof SupportsCredentialVending)) {
+ throw new UnsupportedOperationException(
+ String.format("Filesystem provider for scheme %s cannot use vended
credentials", scheme));
+ }
+
+ PathBasedCredentialContext context =
+ new PathBasedCredentialContext(
+ PrincipalUtils.getCurrentUserName(),
+ Collections.emptySet(),
+ Collections.singleton(path.toString()));
+ Credential credential =
+ manager
+ .getCredential(matchingTypes.get(0), context)
+ .orElseThrow(
+ () ->
+ new ConnectionFailedException(
+ "Credential provider returned no credential for
Fileset connection test"));
+ Credential[] credentials = new Credential[] {credential};
+ String handle =
InMemoryFileSystemCredentialsProvider.register(credentials);
+ try {
+ probeConf.put(
+ GravitinoFileSystemCredentialsProvider.GVFS_CREDENTIAL_PROVIDER,
+ InMemoryFileSystemCredentialsProvider.class.getCanonicalName());
+ probeConf.put(InMemoryFileSystemCredentialsProvider.CREDENTIAL_HANDLE,
handle);
+ probeConf.putAll(
+ ((SupportsCredentialVending) fileSystemProvider)
+ .getFileSystemCredentialConf(credentials));
+ return handle;
+ } catch (RuntimeException e) {
+ InMemoryFileSystemCredentialsProvider.unregister(handle);
+ throw e;
+ }
+ }
+
+ private synchronized CatalogCredentialManager credentialManager() {
Review Comment:
**[Plausible]** This adds its own lazily-created, separately-closed
`CatalogCredentialManager` (built from `conf`), duplicating the
`CatalogCredentialManager` lifecycle that already exists one layer up on
`BaseCatalog` (`core/.../BaseCatalog.java:365-374`, closed at ~336) for the
same catalog.
`FilesetCatalogImpl` already maintains one `CatalogCredentialManager`
instance for GVFS credential vending; this new probe path stands up a second,
independent manager purely for connection-test probes, doubling provider-side
resource usage (e.g. STS/cloud SDK clients) and creating two
independently-evolving code paths for the same concept.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1065,31 +1101,6 @@ private Map<String, Path>
getAndCheckCatalogStorageLocations(Map<String, String>
checkPlaceholderValue(v);
- if (!disableFSOps && !containsPlaceholder(v)) {
- Path path = new Path(v);
- // At catalog initialization, only merge catalog config and
user-defined location
- // configs
- Map<String, String> fsConf = new HashMap<>(conf);
- fsConf.putAll(
- FilesetUtil.getUserDefinedFileSystemConfigs(
- path.toUri(),
- conf,
-
FilesetCatalogPropertiesMetadata.FS_GRAVITINO_PATH_CONFIG_PREFIX));
- FileSystem fs = getFileSystemWithCache(path, fsConf);
- try {
- if (fs.exists(path) && fs.getFileStatus(path).isFile()) {
- throw new IllegalArgumentException(
- "Fileset catalog location cannot be a file: "
Review Comment:
**[Confirmed bug]** This deletes the only eager validation in
`getAndCheckCatalogStorageLocations` (run from `initialize()`, i.e. on every
catalog create/alter/reload) that rejected a location pointing at an existing
regular file, with no replacement in that mandatory path — the equivalent check
now only lives inside the new opt-in `testConnection(NameIdentifier)`.
Creating/altering a Fileset catalog with
`location-archive=file:///tmp/somefile` used to fail immediately with a clear
`IllegalArgumentException`; now it succeeds silently and only surfaces later
via a confusing low-level I/O failure, or never surfaces if the operator
doesn't separately call the new manual `testConnection` endpoint. Confirmed by
this PR's own new test `testTestConnectionRejectsCatalogLocationThatIsAFile`,
which calls `initialize()` with a file location with no `assertThrows` wrapper.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -991,6 +1022,11 @@ public void close() throws IOException {
fileSystemCache.cleanUp();
}
+ if (credentialManager != null) {
Review Comment:
**[Plausible]** `close()` reads/nulls/closes the `credentialManager` field
without synchronization while the lazy-init accessor `credentialManager()` is
`synchronized`, and the field isn't `volatile`.
If one thread is mid-probe inside `addVendedCredential() ->
credentialManager().getCredential(...)` while another thread concurrently calls
`close()` (e.g. catalog being dropped/reloaded), `close()` can close the
manager the first thread is actively using, or fail to observe a manager
created moments earlier by another thread, leaking its underlying credential
provider resources.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1443,226 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ private Map<String, Path> resolveProbeLocations(NameIdentifier catalogIdent)
{
+ Map<String, Path> resolved = new HashMap<>();
+ catalogStorageLocations.forEach(
+ (locationName, location) ->
+ resolved.put(locationName,
resolveProbeLocation(location.toString(), catalogIdent)));
+ return resolved;
+ }
+
+ private Path resolveProbeLocation(String location, NameIdentifier
catalogIdent) {
+ String resolved = location.replace("{{catalog}}", catalogIdent.name());
+ Matcher matcher = LOCATION_PLACEHOLDER_PATTERN.matcher(resolved);
+ if (matcher.find()) {
+ String staticPrefix = resolved.substring(0, matcher.start());
+ int lastSeparator = staticPrefix.lastIndexOf(SLASH);
+ if (lastSeparator < 0) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no concrete parent that can be
tested");
+ }
+ resolved = staticPrefix.substring(0, lastSeparator + 1);
+ }
+
+ try {
Review Comment:
**[Confirmed bug]** `resolveProbeLocation()` rejects syntactically valid
`scheme:///path` URIs with an empty (but legitimate) authority — e.g.
`hdfs:///warehouse` relying on the cluster's `fs.defaultFS` — as "no valid
concrete target that can be tested".
`URI.create("hdfs:///warehouse").getAuthority()` returns `""` (not `null`);
the check `scheme != null && resolved.contains("://") &&
!"file".equalsIgnoreCase(scheme) && StringUtils.isBlank(authority)` evaluates
true and throws `IllegalArgumentException`, propagating uncaught out of
`testConnection()` since locations are resolved eagerly before the per-location
try/catch — failing connection tests for perfectly reachable HDFS catalogs.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -946,24 +958,43 @@ public boolean dropSchema(NameIdentifier ident, boolean
cascade) throws NonEmpty
}
}
- /**
- * Since the Fileset catalog was completely managed by Gravitino, we don't
need to test the
- * connection
- *
- * @param catalogIdent the name of the catalog.
- * @param type the type of the catalog.
- * @param provider the provider of the catalog.
- * @param comment the comment of the catalog.
- * @param properties the properties of the catalog.
- */
@Override
public void testConnection(
NameIdentifier catalogIdent,
Catalog.Type type,
String provider,
String comment,
Map<String, String> properties) {
- // Do nothing
+ testConnection(catalogIdent);
+ }
+
+ @Override
+ public void testConnection(NameIdentifier catalogIdent) {
+ if (disableFSOps) {
+ throw new UnsupportedOperationException(
+ "Fileset connection testing requires filesystem operations to be
enabled");
+ }
+ if (catalogStorageLocations.isEmpty()) {
+ throw new IllegalArgumentException("Fileset catalog has no catalog-level
location to test");
+ }
+
Review Comment:
**[Efficiency]** `testConnection()` probes every catalog-level location
sequentially (blocking on `probeFuture.get(timeout)` inside `probeLocation`
before the `forEach` loop advances) instead of submitting all probes
concurrently to the existing `fileSystemExecutor` thread pool.
With the default `FILESYSTEM_CONNECTION_TIMEOUT_SECONDS` (6s) and a catalog
with N locations, one or more slow/unreachable targets make the whole
`testConnection()` call take up to N × 6s in the worst case, instead of roughly
one timeout period if the N probes were submitted together and each `Future`
awaited afterward.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1443,226 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ private Map<String, Path> resolveProbeLocations(NameIdentifier catalogIdent)
{
+ Map<String, Path> resolved = new HashMap<>();
+ catalogStorageLocations.forEach(
+ (locationName, location) ->
+ resolved.put(locationName,
resolveProbeLocation(location.toString(), catalogIdent)));
+ return resolved;
+ }
+
+ private Path resolveProbeLocation(String location, NameIdentifier
catalogIdent) {
+ String resolved = location.replace("{{catalog}}", catalogIdent.name());
+ Matcher matcher = LOCATION_PLACEHOLDER_PATTERN.matcher(resolved);
+ if (matcher.find()) {
+ String staticPrefix = resolved.substring(0, matcher.start());
+ int lastSeparator = staticPrefix.lastIndexOf(SLASH);
+ if (lastSeparator < 0) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no concrete parent that can be
tested");
+ }
+ resolved = staticPrefix.substring(0, lastSeparator + 1);
+ }
+
+ try {
+ URI uri = URI.create(resolved);
+ if (StringUtils.isBlank(resolved)
+ || containsPlaceholder(resolved)
+ || (uri.getScheme() != null
+ && resolved.contains("://")
+ && !"file".equalsIgnoreCase(uri.getScheme())
+ && StringUtils.isBlank(uri.getAuthority()))) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no valid concrete target that can be
tested");
+ }
+ return new Path(resolved);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no valid concrete target that can be
tested", e);
+ }
+ }
+
+ private void probeLocation(Path path) throws Exception {
Review Comment:
**[Simplification]** `probeLocation()` re-implements, almost line-for-line,
the submit-to-executor / `future.get(timeout)` / `TimeoutException`-cancel /
`InterruptedException` / `ExecutionException`-unwrap scaffolding that
`getFileSystem()` (~line 1367) already has; only the innermost
`FileSystem`-construction lambda was factored into the shared
`createFileSystem()` helper.
The two copies have already drifted (`getFileSystem` logs a WARN with
structured fields on timeout/cancel, while `probeLocation` silently wraps
everything as `IOException` with no log line), so a future fix to one path is
easy to miss applying to the other.
##########
core/src/main/java/org/apache/gravitino/connector/CatalogOperations.java:
##########
@@ -67,4 +67,15 @@ void testConnection(
String comment,
Map<String, String> properties)
throws Exception;
+
+ /**
+ * Test the connection of an existing catalog using its stored configuration.
+ *
+ * @param catalogIdent the identifier of the existing catalog.
+ * @throws Exception if the connection test fails.
+ */
+ default void testConnection(NameIdentifier catalogIdent) throws Exception {
Review Comment:
**[Simplification]** The old 5-arg `testConnection(NameIdentifier,
Catalog.Type, String, String, Map)` stays abstract (no default), forcing all 9
catalog-ops implementations to each add a byte-identical one-line override
`testConnection(catalogIdent);` that just forwards to the new 1-arg method.
The interface already demonstrates the correct pattern one method below (the
new 1-arg `testConnection` has a `default` `UnsupportedOperationException`
body); the same `default` forwarding technique could eliminate all 9 copies.
--
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]
