mchades commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3842649271
##########
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:
Fixed. `testConnection()` now aggregates `UnsupportedOperationException` per
location instead of aborting the remaining probes. Multiple matching credential
providers are reported as `ambiguous credential provider configuration`, while
other unsupported probes are reported as `probe unsupported`. Multi-location
coverage was added.
##########
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:
Agreed. I removed the second `CatalogCredentialManager`.
`FilesetCatalogImpl` now supplies its inherited catalog credential manager to
`FilesetCatalogOperations`, which no longer creates or closes an independent
manager.
##########
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:
Applied. The five-argument method is now a default method that delegates to
`testConnection(NameIdentifier)`, and the redundant forwarding overrides were
removed from the catalog implementations.
##########
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:
Thanks for the suggestion. Multi-location Fileset catalogs are uncommon, and
`testConnection` is not a latency-sensitive path. I kept the probes sequential
in this PR to preserve simpler timeout and resource-cleanup semantics. We can
revisit concurrency if this becomes a practical performance issue.
##########
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:
Good point. I extracted the common submit, wait, cancel, and interrupt
lifecycle into `executeFileSystemTask()`. The callers retain their different
scopes: `getFileSystem()` times client creation, while `probeLocation()` times
client creation plus the metadata request and closes the temporary client.
--
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]
