yuqi1129 commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3850366792
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -951,24 +981,39 @@ 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
+ 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:
Three different error codes are returned for what is really the same
situation: "there is no probe we can run here."
- `disable-filesystem-ops=true` (line 987) returns **1006** unsupported.
- No catalog-level location, this line, returns **1001** illegal argument.
- An unsupported scheme, or a provider that cannot take vended credentials,
ends up aggregated into the `ConnectionFailedException` below and returns
**1007**.
The PR description says:
> If no meaningful catalog-level probe can be executed, the operation is
reported as unsupported.
None of the last two match that. The 1007 case is new in the latest commit:
`testConnection` used to rethrow `UnsupportedOperationException` so it reached
the caller as 1006, and now it is caught by the general `catch (Exception e)`
and turned into a failure message.
`testTestConnectionReportsUnsupportedLocationProbe` asserts the new behaviour,
so it looks intentional, but it does read as a connection failure to the user
when nothing was actually wrong with the connection.
Also worth noting that a catalog with locations only at the schema or
fileset level is a valid configuration, and it now gets a hard 1001 error.
Could all three go through 1006, so "cannot test" and "tested and failed"
stay distinguishable for the caller?
##########
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.
`listStatusIterator` throws `FileNotFoundException` on HDFS and on the local
filesystem when the directory is missing. But a Fileset catalog is allowed to
point at a location that has not been created:
`getAndCheckCatalogStorageLocations` only rejects a path that already exists
**and is a file**, and the directories are created later, when schemas and
filesets are created.
Two things make this bigger than it looks:
1. Because the five-argument `testConnection` now falls through to this one,
the **pre-create** endpoint hits it too. A user checking
`location=hdfs://ns/warehouse/new-catalog` before creating the catalog gets
`1007` for a configuration that would have worked.
2. The result depends on the storage backend. Object stores return an empty
listing for a missing prefix, so the same configuration passes on S3 and fails
on HDFS.
Would it work to treat "not found" on the catalog root as a pass, or to walk
up to the nearest existing parent and probe that instead?
##########
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, and for a typical object-store layout that is the bucket
root.
For `location = s3://bucket/{{schema}}/data` the static prefix is
`s3://bucket/`, `lastSeparator` is 11 and `schemeDelimiter + 3` is 5, so the
new guard passes and the probe targets `s3://bucket/`.
That has two effects:
- Deployments that scope IAM to a sub-prefix get `access denied` for a
catalog that is perfectly healthy, because listing the bucket root is not
allowed.
- The credential context on line 1617 is built from the same truncated path,
so the probe also **requests credentials for a wider path** than the catalog
ever uses.
The check added in the last commit fixes the `s3://{{schema}}-bucket/x`
case, which is good, but the common case above is unchanged. If probing the
real prefix is not possible, it would help to say in the error message which
path was actually tested, so the user is not misled.
##########
core/src/main/java/org/apache/gravitino/connector/CatalogOperations.java:
##########
@@ -60,11 +60,24 @@ void initialize(
* @param properties the properties of the catalog.
* @throws Exception if connection fails.
*/
- void testConnection(
+ default void testConnection(
Review Comment:
The PR description makes a compatibility claim about this default method:
> The new Java and connector SPI methods use default implementations, so
this does not introduce a binary or source incompatibility for existing
implementations.
I could not find a test for it. I searched the test sources and every test
either implements the one-argument `testConnection` or calls it directly;
nothing implements **only** the five-argument version, which is what an
existing third-party `CatalogOperations` would look like.
That case is worth one small test, because it is the whole point of the
default: an implementation that overrides only the five-argument method should
still work on the pre-create endpoint, and should return 1006 on the new
endpoint.
One small correction to the same sentence:
`core/src/main/java/org/apache/gravitino/catalog/SupportsCatalogs` gained `void
testConnection(NameIdentifier ident)` as an **abstract** method, not a default
one. The public `api` interface does use a default, so the claim holds there,
but it is not true for all the interfaces this PR touches.
##########
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:
A few smaller things in this area.
**`toLowerCase()` without a locale** (this line and the next). The rest of
the codebase uses `Locale.ROOT` for this, and under a Turkish locale `I`
lowercases to `ı`, so a message-based check can silently stop matching.
**Matching on messages this class itself produced.** `message.contains("must
be a directory")` and the `no concrete parent` / `no valid concrete target`
checks are recognising strings thrown a few lines above. Using a small private
exception type (or checking the type directly) would not break when someone
edits the wording. The `multiple credential providers` case was already cleaned
up this way, so this is the same idea applied to the rest.
**Credential context is not filtered by provider** (line 1620). The
production path, `CatalogCredentialManager.getCredentialByPath`, runs
`CredentialOperationDispatcher.filterContextByProvider` first, which drops
paths the provider does not support via `isPathSupported`. The probe calls
`manager.getCredential(type, context)` directly, so a provider that supports
the scheme but not this particular path is skipped in normal use and invoked
here.
**Credential handle is removed while a timed-out task may still be
running.** On timeout, `executeFileSystemTask` cancels by interrupt, but the
worker may not stop immediately, and the `finally` in `probeLocation`
unregisters the handle right away. The orphaned task then fails with
"Credentials are no longer available", which is confusing because it has
nothing to do with the real problem.
--
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]
