mchades commented on code in PR #12553:
URL: https://github.com/apache/gravitino/pull/12553#discussion_r3842647951
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1443,226 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ 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;
+ }
+
+ private Path resolveProbeLocation(String location, NameIdentifier
catalogIdent) {
+ String resolved = location.replace("{{catalog}}", catalogIdent.name());
Review Comment:
Fixed in `00a205e497`. I extracted the token into a private
`CATALOG_PLACEHOLDER` constant and now use it in `resolveProbeLocation()`.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1443,226 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ 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;
+ }
+
+ private Path resolveProbeLocation(String location, NameIdentifier
catalogIdent) {
+ String resolved = location.replace("{{catalog}}", 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 {
+ URI uri = URI.create(resolved);
+ if (StringUtils.isBlank(resolved)
+ || containsPlaceholder(resolved)
+ || (uri.getScheme() != null
+ && resolved.contains("://")
+ && !"file".equalsIgnoreCase(uri.getScheme())
+ && StringUtils.isBlank(uri.getAuthority()))) {
+ 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;
+ Future<Void> probeFuture = null;
+ try {
+ credentialHandle = addVendedCredential(path, scheme, fileSystemProvider,
probeConf);
+ int timeoutSeconds =
+ (int)
+ propertiesMetadata
+ .catalogPropertiesMetadata()
+ .getOrDefault(
+ probeConf,
+
FilesetCatalogPropertiesMetadata.FILESYSTEM_CONNECTION_TIMEOUT_SECONDS);
+ probeFuture =
+ fileSystemExecutor.submit(
+ () -> {
+ 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;
+ }
+ });
+ probeFuture.get(timeoutSeconds, TimeUnit.SECONDS);
+ } catch (TimeoutException e) {
+ if (probeFuture != null) {
+ probeFuture.cancel(true);
+ }
+ throw new IOException("Filesystem connection probe timed out", e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ 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(
Review Comment:
Yes. I added adapter tests for S3, OSS, ABS, GCS, and COS to verify that
vended credentials are translated into the expected HCFS credential
configuration. I also added and ran `FilesetS3TokenConnectionIT` against MinIO
to cover S3 token vending and the Fileset metadata probe end to end. All passed
locally.
##########
server/src/main/java/org/apache/gravitino/server/web/rest/ExceptionHandlers.java:
##########
@@ -212,6 +199,38 @@ public static Response handlePartitionStatsException(
return PartitionStatsExceptionHandler.INSTANCE.handle(type, name, parent,
e);
}
+ private static Response handleTestConnectionException(Exception e, boolean
includeStack) {
+ Throwable throwable = includeStack ? e : null;
+ ErrorResponse response;
+ if (e instanceof IllegalArgumentException) {
+ response = ErrorResponse.illegalArguments(e.getMessage(), throwable);
+
+ } else if (e instanceof ConnectionFailedException) {
+ response = ErrorResponse.connectionFailed(e.getMessage(), throwable);
+
+ } else if (e instanceof UnsupportedOperationException) {
+ response = ErrorResponse.unsupportedOperation(e.getMessage(), throwable);
+
+ } else if (e instanceof NotFoundException) {
+ response = ErrorResponse.notFound(e.getClass().getSimpleName(),
e.getMessage(), throwable);
+
+ } else if (e instanceof AlreadyExistsException) {
+ response =
+ ErrorResponse.alreadyExists(e.getClass().getSimpleName(),
e.getMessage(), throwable);
+
+ } else if (e instanceof NotInUseException) {
+ response = ErrorResponse.notInUse(e.getClass().getSimpleName(),
e.getMessage(), throwable);
+
+ } else {
+ return Utils.internalError(e.getMessage(), e);
+ }
+
Review Comment:
Fixed in `00a205e497`. The fallback now passes the sanitized `throwable`
instead of the raw exception. I also added REST coverage for an unexpected
`RuntimeException`, verifying that it returns HTTP 500 without exposing the
stack trace.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1065,31 +1101,6 @@ private Map<String, Path>
getAndCheckCatalogStorageLocations(Map<String, String>
checkPlaceholderValue(v);
- if (!disableFSOps && !containsPlaceholder(v)) {
- Path path = new Path(v);
- // At catalog initialization, only merge catalog config and
user-defined location
- // configs
- Map<String, String> fsConf = new HashMap<>(conf);
- fsConf.putAll(
- FilesetUtil.getUserDefinedFileSystemConfigs(
- path.toUri(),
- conf,
-
FilesetCatalogPropertiesMetadata.FS_GRAVITINO_PATH_CONFIG_PREFIX));
- FileSystem fs = getFileSystemWithCache(path, fsConf);
- try {
- if (fs.exists(path) && fs.getFileStatus(path).isFile()) {
- throw new IllegalArgumentException(
- "Fileset catalog location cannot be a file: "
Review Comment:
Agreed. I restored the original initialization-time regular-file validation
for concrete named `location-*` entries when filesystem operations are enabled.
The explicit `testConnection` probe also covers the default `location`, which
was not covered by the legacy check. Tests were added for both behaviors.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -1441,6 +1443,226 @@ FileSystem getFileSystem(Path path, Map<String, String>
config) throws IOExcepti
}
}
+ 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;
+ }
+
+ private Path resolveProbeLocation(String location, NameIdentifier
catalogIdent) {
+ String resolved = location.replace("{{catalog}}", 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 {
Review Comment:
Fixed. `resolveProbeLocation()` no longer rejects a URI solely because its
authority is empty. A location such as `hdfs:///warehouse` is now passed to
Hadoop so it can be resolved using `fs.defaultFS`. I added a unit test for this
case.
##########
catalogs/catalog-fileset/src/main/java/org/apache/gravitino/catalog/fileset/FilesetCatalogOperations.java:
##########
@@ -991,6 +1022,11 @@ public void close() throws IOException {
fileSystemCache.cleanUp();
}
+ if (credentialManager != null) {
Review Comment:
Fixed by adding a catalog tree read lock around existing-catalog
`testConnection`, so drop or alter cannot close the catalog while the probe is
running. The operations-local credential manager was also removed in favor of
the existing `BaseCatalog` manager. I added a deterministic lock test.
--
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]
