mchades commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3851435527
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1446,6 +1477,229 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ private <T> T executeFileSystemTask(Callable<T> task, int timeoutSeconds)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ Future<T> future = fileSystemExecutor.submit(task);
+ try {
+ return future.get(timeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ throw e;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ }
+
+ @VisibleForTesting
+ Path resolveProbeLocation(String location, NameIdentifier catalogIdent) {
+ String resolved = location.replace(CATALOG_PLACEHOLDER,
catalogIdent.name());
+ Matcher matcher = LOCATION_PLACEHOLDER_PATTERN.matcher(resolved);
+ if (matcher.find()) {
+ String staticPrefix = resolved.substring(0, matcher.start());
+ int lastSeparator = staticPrefix.lastIndexOf(SLASH);
+ int schemeDelimiter = staticPrefix.indexOf(SCHEME_DELIMITER);
+ if (lastSeparator < 0
+ || (schemeDelimiter >= 0
+ && lastSeparator < schemeDelimiter + SCHEME_DELIMITER.length()))
{
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no concrete parent that can be
tested");
+ }
+ resolved = staticPrefix.substring(0, lastSeparator + 1);
+ }
+
+ try {
+ if (StringUtils.isBlank(resolved) || containsPlaceholder(resolved)) {
+ 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;
+ try {
+ credentialHandle = addVendedCredential(path, scheme, fileSystemProvider,
probeConf);
+ int timeoutSeconds =
+ (int)
+ propertiesMetadata
+ .catalogPropertiesMetadata()
+ .getOrDefault(
+ probeConf,
+
FilesetCatalogPropertiesMetadata.FILESYSTEM_CONNECTION_TIMEOUT_SECONDS);
+ executeFileSystemTask(
+ () -> {
+ try (FileSystem fileSystem =
+ createFileSystem(path, probeConf, scheme, fileSystemProvider))
{
+ Path qualifiedPath =
+ path.makeQualified(fileSystem.getUri(),
fileSystem.getWorkingDirectory());
+ RemoteIterator<FileStatus> iterator =
fileSystem.listStatusIterator(qualifiedPath);
Review Comment:
> A catalog location that does not exist yet is reported as a connection
failure.
Fixed in `7eafb37fc1`. A `FileNotFoundException` when listing the catalog
location is now treated as a successful probe because the location may be
created later by schema or fileset operations. Other I/O, authorization, and
connectivity failures are still reported. Tests were added for missing default
and named locations.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1446,6 +1477,229 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ private <T> T executeFileSystemTask(Callable<T> task, int timeoutSeconds)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ Future<T> future = fileSystemExecutor.submit(task);
+ try {
+ return future.get(timeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ throw e;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ }
+
+ @VisibleForTesting
+ Path resolveProbeLocation(String location, NameIdentifier catalogIdent) {
+ String resolved = location.replace(CATALOG_PLACEHOLDER,
catalogIdent.name());
+ Matcher matcher = LOCATION_PLACEHOLDER_PATTERN.matcher(resolved);
+ if (matcher.find()) {
+ String staticPrefix = resolved.substring(0, matcher.start());
+ int lastSeparator = staticPrefix.lastIndexOf(SLASH);
+ int schemeDelimiter = staticPrefix.indexOf(SCHEME_DELIMITER);
+ if (lastSeparator < 0
+ || (schemeDelimiter >= 0
+ && lastSeparator < schemeDelimiter + SCHEME_DELIMITER.length()))
{
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no concrete parent that can be
tested");
+ }
+ resolved = staticPrefix.substring(0, lastSeparator + 1);
Review Comment:
> When the location contains a placeholder, the probe falls back to the
nearest static parent...
Fixed in `7eafb37fc1`. The probe now substitutes only `{{catalog}}`. If any
placeholder remains unresolved, it reports `probe unsupported` without
requesting credentials or performing filesystem I/O. The failure message
identifies the unresolved placeholder and resolved location, and states that no
filesystem path was tested. Locations containing only `{{catalog}}` continue to
probe the exact resolved path.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1446,6 +1477,229 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ private <T> T executeFileSystemTask(Callable<T> task, int timeoutSeconds)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ Future<T> future = fileSystemExecutor.submit(task);
+ try {
+ return future.get(timeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ future.cancel(true);
+ throw e;
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ }
+
+ @VisibleForTesting
+ Path resolveProbeLocation(String location, NameIdentifier catalogIdent) {
+ String resolved = location.replace(CATALOG_PLACEHOLDER,
catalogIdent.name());
+ Matcher matcher = LOCATION_PLACEHOLDER_PATTERN.matcher(resolved);
+ if (matcher.find()) {
+ String staticPrefix = resolved.substring(0, matcher.start());
+ int lastSeparator = staticPrefix.lastIndexOf(SLASH);
+ int schemeDelimiter = staticPrefix.indexOf(SCHEME_DELIMITER);
+ if (lastSeparator < 0
+ || (schemeDelimiter >= 0
+ && lastSeparator < schemeDelimiter + SCHEME_DELIMITER.length()))
{
+ throw new IllegalArgumentException(
+ "Fileset catalog location has no concrete parent that can be
tested");
+ }
+ resolved = staticPrefix.substring(0, lastSeparator + 1);
+ }
+
+ try {
+ if (StringUtils.isBlank(resolved) || containsPlaceholder(resolved)) {
+ 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;
+ try {
+ credentialHandle = addVendedCredential(path, scheme, fileSystemProvider,
probeConf);
+ int timeoutSeconds =
+ (int)
+ propertiesMetadata
+ .catalogPropertiesMetadata()
+ .getOrDefault(
+ probeConf,
+
FilesetCatalogPropertiesMetadata.FILESYSTEM_CONNECTION_TIMEOUT_SECONDS);
+ executeFileSystemTask(
+ () -> {
+ 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;
+ }
+ },
+ timeoutSeconds);
+ } catch (TimeoutException e) {
+ throw new IOException("Filesystem connection probe timed out", e);
+ } catch (InterruptedException e) {
+ 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 = catalogCredentialManager();
+ 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 (!(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[] credentials =
+ matchingTypes.stream()
+ .map(type -> manager.getCredential(type, context))
+ .filter(Optional::isPresent)
+ .map(Optional::get)
+ .toArray(Credential[]::new);
+ if (credentials.length == 0) {
+ throw new ConnectionFailedException(
+ "Credential providers returned no credential for Fileset connection
test");
+ }
+ 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;
+ }
+ }
+
+ @VisibleForTesting
+ CatalogCredentialManager catalogCredentialManager() {
+ Preconditions.checkState(
+ credentialManagerSupplier != null,
+ "Catalog credential manager is not available for Fileset connection
testing");
+ return Objects.requireNonNull(
+ credentialManagerSupplier.get(), "Catalog credential manager must not
be null");
+ }
+
+ private FileSystem createFileSystem(
+ Path path, Map<String, String> config, String scheme, FileSystemProvider
provider)
+ throws IOException {
+ if (scheme.equals(SCHEME_HDFS)) {
+ return new ImpersonationHDFSFileSystemProxy(path, config,
PrincipalUtils::getCurrentUserName)
+ .getProxy();
+ }
+ return provider.getFileSystem(path, config);
+ }
+
+ private String displayLocationName(String locationName) {
+ return LOCATION_NAME_UNKNOWN.equals(locationName) ? "location" :
"location-" + locationName;
+ }
+
+ private String failureCategory(Throwable throwable) {
+ Throwable current = throwable;
+ while (current != null) {
+ String simpleName = current.getClass().getSimpleName().toLowerCase();
Review Comment:
> **`toLowerCase()` without a locale** ...
Fixed in `7eafb37fc1`. Failure classification now uses `Locale.ROOT`.
> **Matching on messages this class itself produced.** ...
Fixed in the same commit. Internally generated probe failures now use
private exception types instead of relying on their message text. Message
inspection remains only for provider and filesystem failures whose exception
types are not consistent across backends.
--
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]
