mchades commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3850996249
##########
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:
On the credential-context point, filterContextByProvider() currently filters
PathBasedCredentialContext only by URI scheme through supportsScheme(); it does
not apply a finer-grained path check. This probe already filters each
credential provider with the same supportsScheme(scheme) check, and its context
contains only the single probed path, so applying that filter would not change
the context. getCredentialByPath() is also not a direct replacement because it
rejects multiple matching providers, while the probe intentionally supports all
configured providers matching the scheme. I am therefore keeping this part
unchanged.
--
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]
