jerryshao commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3843096618
##########
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);
Review Comment:
**[Confirmed bug]** `resolveProbeLocations(catalogIdent)` is called outside
the per-location try/catch, so a single unresolvable location aborts the entire
`testConnection` call instead of being aggregated with other location failures.
`resolveProbeLocations` eagerly resolves every configured location up front,
before the try/catch in the `forEach` loop even starts. `resolveProbeLocation`
throws `IllegalArgumentException` for a malformed `location-*` placeholder
(e.g. "no concrete parent that can be tested"). This exception propagates
unhandled straight out of `testConnection()` — no locations get probed at all,
and the `failures` aggregation list (verified by the PR's own
`testTestConnectionAggregatesLocationFailures` test) never runs, so one bad
location property silently prevents even other perfectly valid, reachable
locations from being tested or reported together.
##########
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:
**[Confirmed bug]** `testConnection` only catches
`software.amazon.awssdk.services.glue.model.GlueException`, missing the sibling
`SdkClientException` thrown for unreachable-endpoint/invalid-credential
failures.
`GlueException` and `SdkClientException` are both subclasses of
`SdkException` but are siblings, not parent/child. Testing an existing Glue
catalog with invalid AWS credentials or an unreachable endpoint causes
`glueClient.getDatabases()` to throw `SdkClientException`, which is not caught
here, so it propagates as an unmapped `RuntimeException`.
`ExceptionHandlers.handleTestConnectionException`'s catch-all then falls
through to `Utils.internalError` (HTTP 500) instead of the documented 200 +
`ConnectionFailedException` (1007) response that Hive/Iceberg/Hudi/Kafka
produce via their broader `catch (Exception e)` for the exact class of failure
(connectivity) a connection test exists to surface.
##########
clients/client-python/gravitino/client/gravitino_metalake.py:
##########
@@ -351,6 +351,25 @@ def disable_catalog(self, name: str):
url, json=catalog_disable_request,
error_handler=CATALOG_ERROR_HANDLER
)
+ def test_connection(self, name: str) -> None:
+ """Test an existing catalog connection using its stored configuration.
+
+ Args:
+ name: The name of the existing catalog.
+
+ Raises:
+ NoSuchCatalogException: If the catalog does not exist.
+ UnsupportedOperationException: If the catalog does not define a
connection probe.
+ ConnectionFailedException: If the catalog cannot reach its
external system.
+ """
+ url = (
+ self.API_METALAKES_CATALOGS_PATH.format(
+ encode_string(self.name()), encode_string(name)
+ )
+ + "/testConnection"
+ )
+ self.rest_client.post(url, error_handler=CATALOG_ERROR_HANDLER)
Review Comment:
**[Confirmed bug]** `test_connection()` never checks the response body's
`code` field, so it silently returns success even when the connection test
actually failed.
The server's `handleTestConnectionException` always returns HTTP 200
(`Response.status(Response.Status.OK)`) with the failure encoded in the JSON
body's `code` field for `ConnectionFailedException`,
`UnsupportedOperationException`, `IllegalArgumentException`, etc. The Python
`HttpClient`'s `_make_request` determines success purely from HTTP status, so a
200 response is always treated as success. `test_connection` just calls
`self.rest_client.post(url, error_handler=CATALOG_ERROR_HANDLER)` and returns
`None` without inspecting the body, unlike Java's
`GravitinoMetalake.testConnection(String)` which explicitly checks
`resp.getCode() == 0` and invokes the error handler otherwise. A Python user
testing a catalog with bad credentials or an unreachable endpoint gets no
exception at all.
##########
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:
**[Plausible]** `addVendedCredential` throws `UnsupportedOperationException`
when more than one configured credential provider supports a probed location's
scheme, unlike the production credential-vending path which merges credentials
from multiple matching providers without complaint.
`CredentialOperationDispatcher.getCredentials`
(core/.../credential/CredentialOperationDispatcher.java) iterates all
configured providers whose scheme matches and merges their credentials with no
ambiguity check — this is what real file reads/writes use. Multiple built-in
AWS providers (`S3TokenProvider`, `S3SecretKeyProvider`,
`AwsIrsaCredentialProvider`) legitimately overlap on scheme s3a/s3/s3n, and
`CredentialConfig.CREDENTIAL_PROVIDERS` allows any comma-separated list with no
overlap validation. A Fileset catalog configured with two same-scheme providers
— a working, supported configuration for actual file access — gets a false
"ambiguous credential provider configuration" failure from `testConnection`
even though the catalog functions correctly in production.
##########
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:
**[Plausible]** `resolveProbeLocation`'s placeholder-truncation logic
silently drops the bucket/authority when a location is missing the separating
`/` before a placeholder, instead of raising a clear error.
For location `s3://bucket{{schema}}` (missing `/` before the placeholder),
`staticPrefix` becomes `s3://bucket`; `lastIndexOf('/')` finds the slash inside
the scheme's `//` (index 4) rather than a real path separator, truncating the
result to bare `s3://`. The blank/placeholder guard doesn't catch this case,
and `new Path("s3://")` succeeds silently with the bucket name entirely lost,
so the probe runs against an unintended, invalid target instead of surfacing
the misconfiguration with a clear "no valid concrete target" error.
##########
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:
**[Plausible]** This per-location `catch (Exception e)` block swallows
exceptions with no logging anywhere in the entire call chain, turning genuine
bugs (e.g. an NPE in `probeLocation`/`addVendedCredential`) into an
undiagnosable generic "probe failed" category.
`catch (Exception e) { failures.add(... + failureCategory(e) + ...) }` never
logs `e`. `failureCategory()` only pattern-matches the message/class name to
classify — it does not log either. The REST handler's `LOG.info` call for this
path only logs the metalake/catalog name (2-arg call), never the exception. An
unexpected coding bug deep in the new probe/credential-vending logic is
reported to the caller as an opaque "probe failed" with zero trace anywhere in
server logs, making the new code path effectively undebuggable in production.
##########
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:
**[Plausible, side effect elsewhere in this file]** Not about this new
method directly, but a consequence of this PR's design change: the *pre-create*
`testConnection(ident, type, provider, comment, properties)` method above
(unchanged by this diff, `catch (Exception e)` block at line 742 doing
`LOG.warn("Failed to test catalog creation {}", ident, e)`) now fires on every
Model/Generic catalog "Test Connection" click.
Per this PR's own compatibility note, Model/Generic catalogs intentionally
now throw `UnsupportedOperationException` from the pre-create test-connection
path (previously a silent no-op). Since `UnsupportedOperationException` is not
a `GravitinoRuntimeException`, that pre-existing generic catch branch logs a
full WARN-level stack trace for this new, expected, documented case — adding
non-actionable noise to log-monitoring/alerting pipelines. Worth either
narrowing that catch or short-circuiting `UnsupportedOperationException` before
it reaches the WARN log.
##########
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(
Review Comment:
**[Plausible efficiency]** `testConnection` probes each configured location
sequentially rather than concurrently, even though the class already has a
`fileSystemExecutor` thread pool that could run independent per-location probes
in parallel.
`probeLocations.forEach` calls `probeLocation(path)` one at a time; each
call submits to `fileSystemExecutor` internally but blocks synchronously on
`future.get(timeout)` before the loop proceeds to the next location. For a
catalog with N configured locations pointing at slow/unreachable endpoints,
wall-clock time for `testConnection` is the SUM of each location's timeout
rather than the MAX, all held on the synchronous REST request thread.
##########
catalogs/catalog-kafka/src/main/java/org/apache/gravitino/catalog/kafka/KafkaCatalogOperations.java:
##########
@@ -185,12 +184,7 @@ public NameIdentifier[] listTopics(Namespace namespace)
throws NoSuchSchemaExcep
}
@Override
- public void testConnection(
- NameIdentifier catalogIdent,
- Catalog.Type type,
- String provider,
- String comment,
- Map<String, String> properties) {
+ public void testConnection(NameIdentifier catalogIdent) {
try {
Review Comment:
**[Plausible]** `testConnection` calls
`adminClient.listTopics().names().get()` with no explicit timeout, relying on
`AdminClient`'s default 60s `default.api.timeout.ms`.
The `AdminClientConfig` built for this class only sets
`BOOTSTRAP_SERVERS_CONFIG` and `CLIENT_ID_CONFIG`, with no request/timeout
tuning for a fast-fail test. Testing an existing Kafka catalog against an
unreachable broker blocks the synchronous REST thread handling `POST
/testConnection` for up to 60 seconds, unlike the Fileset catalog's
`testConnection` which enforces an explicit short per-location timeout.
--
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]
