mchades commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3844674594
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java:
##########
@@ -155,13 +154,7 @@ public void initialize(
}
@Override
- public void testConnection(
- NameIdentifier catalogIdent,
- Catalog.Type type,
- String provider,
- String comment,
- Map<String, String> properties)
- throws Exception {
+ public void testConnection(NameIdentifier catalogIdent) {
try {
Review Comment:
Fixed in `65da21fe61`. `testConnection` now catches the common
`SdkException` base class, covering both service-side `GlueException` and
client-side `SdkClientException` failures. I added a focused
`SdkClientException` test that verifies the original cause is preserved.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1463,236 @@ 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;
+ }
+ }
+
+ 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;
+ }
+
+ @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);
+ if (lastSeparator < 0) {
+ 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 =
Review Comment:
Fixed in `65da21fe61`. The probe now collects credentials from every
matching provider, consistent with the production credential-vending path. When
providers match, it fails only if none returns a credential. I added a test
verifying that credentials from two matching providers reach the HCFS provider.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1463,236 @@ 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;
+ }
+ }
+
+ 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;
+ }
+
+ @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);
Review Comment:
Fixed in `65da21fe61`. The placeholder boundary check now distinguishes the
URI scheme delimiter from a real path separator, so `s3://bucket{{schema}}` is
rejected as invalid configuration instead of being reduced to `s3://`. A
focused unit test was added.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -946,24 +975,31 @@ 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");
+ }
+
+ Map<String, Path> probeLocations = resolveProbeLocations(catalogIdent);
+ List<String> failures = new ArrayList<>();
+ probeLocations.forEach(
+ (locationName, path) -> {
+ try {
Review Comment:
Fixed in `65da21fe61`. Each per-location failure now logs the catalog,
logical location name, failure category, and full exception stack at WARN
level. The aggregated API response remains sanitized and does not expose the
raw location.
##########
core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java:
##########
@@ -750,6 +750,34 @@ public void testConnection(
}
}
+ /**
+ * Test the connection of an existing catalog using its stored configuration.
+ *
+ * @param ident The identifier of the existing catalog.
+ */
+ @Override
+ public void testConnection(NameIdentifier ident) {
+ TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.READ,
Review Comment:
Fixed in `65da21fe61`. Pre-creation `UnsupportedOperationException` is now
rethrown before the generic WARN branch, so the expected Model/Generic
unsupported result no longer produces a stack trace.
##########
core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java:
##########
@@ -750,6 +750,34 @@ public void testConnection(
}
}
+ /**
+ * Test the connection of an existing catalog using its stored configuration.
+ *
+ * @param ident The identifier of the existing catalog.
+ */
+ @Override
+ public void testConnection(NameIdentifier ident) {
+ TreeLockUtils.doWithTreeLock(
+ ident,
+ LockType.READ,
+ () -> {
+ CatalogWrapper wrapper = loadCatalogAndWrap(ident);
+ wrapper.catalog().checkMetalakeAndCatalogInUse();
+ try {
+ wrapper.doWithCatalogOps(
+ c -> {
+ c.testConnection(ident);
+ return null;
+ });
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(e);
Review Comment:
Fixed in `65da21fe61`. Failures from the existing-catalog path are now
logged with their stack trace on the server before the runtime exception is
preserved or wrapped. Expected `UnsupportedOperationException` remains quiet,
and the client response still omits the stack.
##########
clients/client-java/src/main/java/org/apache/gravitino/client/ErrorHandlers.java:
##########
@@ -562,6 +562,9 @@ public void accept(ErrorResponse errorResponse) {
case ErrorConstants.CONNECTION_FAILED_CODE:
throw new ConnectionFailedException(errorMessage);
+ case ErrorConstants.UNSUPPORTED_OPERATION_CODE:
Review Comment:
Yes. Added in `65da21fe61`. The Python client now parses the
application-level response code and routes failures through
`CATALOG_ERROR_HANDLER`, with tests for both connection failure and unsupported
operation.
##########
catalogs/catalog-glue/src/main/java/org/apache/gravitino/catalog/glue/GlueCatalogOperations.java:
##########
@@ -155,13 +154,7 @@ public void initialize(
}
@Override
- public void testConnection(
- NameIdentifier catalogIdent,
- Catalog.Type type,
- String provider,
- String comment,
- Map<String, String> properties)
- throws Exception {
+ public void testConnection(NameIdentifier catalogIdent) {
Review Comment:
Addressed together with [the earlier Glue
thread](https://github.com/apache/gravitino/pull/12553#discussion_r3843096621)
in `65da21fe61`. The catch now uses `SdkException`, and a focused
`SdkClientException` test verifies that client-side failures become
`ConnectionFailedException`.
--
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]
