This is an automated email from the ASF dual-hosted git repository.
jshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 0e693f77b7 [#6939] feat(GVFS): make Java GVFS operations customizable
(#6961)
0e693f77b7 is described below
commit 0e693f77b7ad9ed0da27c2e459dce030bbde47db
Author: mchades <[email protected]>
AuthorDate: Wed Apr 16 16:15:40 2025 +0800
[#6939] feat(GVFS): make Java GVFS operations customizable (#6961)
### What changes were proposed in this pull request?
- add an abstract class for GVFS ops
- make GVFS ops customizable
- add filesetCatalogCache for cache reuse
### Why are the changes needed?
Fix: #6939
### Does this PR introduce _any_ user-facing change?
yes, developer can customize their own fs ops
### How was this patch tested?
CI pass
---
.../filesystem/hadoop/BaseGVFSOperations.java | 782 +++++++++++++++++++++
.../filesystem/hadoop/DefaultGVFSOperations.java | 234 ++++++
.../filesystem/hadoop/FilesetCatalogCache.java | 100 +++
.../hadoop/FilesetPathNotFoundException.java | 11 +
.../hadoop/GravitinoVirtualFileSystem.java | 624 +++-------------
.../GravitinoVirtualFileSystemConfiguration.java | 7 +
.../filesystem/hadoop/GravitinoMockServerBase.java | 14 +-
.../gravitino/filesystem/hadoop/TestGvfsBase.java | 199 +++++-
docs/how-to-use-gvfs.md | 39 +-
9 files changed, 1428 insertions(+), 582 deletions(-)
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java
new file mode 100644
index 0000000000..a79af67a23
--- /dev/null
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java
@@ -0,0 +1,782 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.filesystem.hadoop;
+
+import static org.apache.gravitino.file.Fileset.PROPERTY_DEFAULT_LOCATION_NAME;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CURRENT_LOCATION_NAME;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.extractIdentifier;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.getConfigMap;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.getSubPathFromGvfsPath;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import com.google.common.collect.Streams;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import java.util.Set;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.audit.CallerContext;
+import org.apache.gravitino.audit.FilesetAuditConstants;
+import org.apache.gravitino.audit.FilesetDataOperation;
+import org.apache.gravitino.audit.InternalClientType;
+import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
+import
org.apache.gravitino.catalog.hadoop.fs.GravitinoFileSystemCredentialsProvider;
+import org.apache.gravitino.catalog.hadoop.fs.SupportsCredentialVending;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.exceptions.CatalogNotInUseException;
+import org.apache.gravitino.exceptions.GravitinoRuntimeException;
+import org.apache.gravitino.exceptions.NoSuchCatalogException;
+import org.apache.gravitino.exceptions.NoSuchFilesetException;
+import org.apache.gravitino.exceptions.NoSuchLocationNameException;
+import org.apache.gravitino.file.Fileset;
+import org.apache.gravitino.file.FilesetCatalog;
+import org.apache.gravitino.storage.AzureProperties;
+import org.apache.gravitino.storage.OSSProperties;
+import org.apache.gravitino.storage.S3Properties;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.Progressable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Abstract base class for Gravitino Virtual File System operations.
+ *
+ * <p>This class provides the core functionality for interacting with the
Gravitino Virtual File
+ * System, including operations for file manipulation, directory management,
and credential
+ * handling. It handles the mapping between virtual paths in the Gravitino
namespace and actual file
+ * paths in underlying storage systems.
+ *
+ * <p>Implementations of this class should provide the specific behaviors for
each file system
+ * operation.
+ */
+public abstract class BaseGVFSOperations implements Closeable {
+ private static final Logger LOG =
LoggerFactory.getLogger(BaseGVFSOperations.class);
+ private static final String SLASH = "/";
+ private static final Set<String> CATALOG_NECESSARY_PROPERTIES_TO_KEEP =
+ Sets.newHashSet(
+ OSSProperties.GRAVITINO_OSS_ENDPOINT,
+ OSSProperties.GRAVITINO_OSS_REGION,
+ S3Properties.GRAVITINO_S3_ENDPOINT,
+ S3Properties.GRAVITINO_S3_REGION,
+ AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME);
+
+ private final String metalakeName;
+
+ private final FilesetCatalogCache filesetCatalogCache;
+
+ private final Configuration conf;
+
+ // Since Caffeine does not ensure that removalListener will be involved
after expiration
+ // We use a scheduler with one thread to clean up expired clients.
+ private final ScheduledThreadPoolExecutor internalFileSystemCleanScheduler;
+
+ // Fileset nameIdentifier-locationName Pair and its corresponding FileSystem
cache, the name
+ // identifier has four levels, the first level is metalake name.
+ private final Cache<Pair<NameIdentifier, String>, FileSystem>
internalFileSystemCache;
+
+ private final Map<String, FileSystemProvider> fileSystemProvidersMap;
+
+ private final String currentLocationEnvVar;
+
+ @Nullable private final String currentLocationName;
+
+ private final long defaultBlockSize;
+
+ /**
+ * Constructs a new {@link BaseGVFSOperations} with the given {@link
Configuration}.
+ *
+ * @param configuration the configuration
+ */
+ public BaseGVFSOperations(Configuration configuration) {
+ this.metalakeName =
+
configuration.get(GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_METALAKE_KEY);
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(metalakeName),
+ "'%s' is not set in the configuration",
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_METALAKE_KEY);
+
+ GravitinoClient client =
GravitinoVirtualFileSystemUtils.createClient(configuration);
+ this.filesetCatalogCache = new FilesetCatalogCache(client);
+
+ this.internalFileSystemCleanScheduler =
+ new ScheduledThreadPoolExecutor(1,
newDaemonThreadFactory("gvfs-filesystem-cache-cleaner"));
+ this.internalFileSystemCache =
+ newFileSystemCache(configuration, internalFileSystemCleanScheduler);
+
+ this.fileSystemProvidersMap =
ImmutableMap.copyOf(getFileSystemProviders());
+
+ this.currentLocationEnvVar =
+ configuration.get(
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CURRENT_LOCATION_NAME_ENV_VAR,
+ GravitinoVirtualFileSystemConfiguration
+ .FS_GRAVITINO_CURRENT_LOCATION_NAME_ENV_VAR_DEFAULT);
+ this.currentLocationName = initCurrentLocationName(configuration);
+
+ this.defaultBlockSize =
+ configuration.getLong(
+ GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE,
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE_DEFAULT);
+
+ this.conf = configuration;
+ }
+
+ @Override
+ public void close() throws IOException {
+ // close all actual FileSystems
+ for (FileSystem fileSystem : internalFileSystemCache.asMap().values()) {
+ try {
+ fileSystem.close();
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ internalFileSystemCache.invalidateAll();
+ internalFileSystemCleanScheduler.shutdownNow();
+
+ try {
+ if (filesetCatalogCache != null) {
+ filesetCatalogCache.close();
+ }
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+
+ /**
+ * Open a file for reading. Same as {@link FileSystem#open(Path, int)}.
+ *
+ * @param gvfsPath the virtual path of the file.
+ * @param bufferSize the buffer size.
+ * @return the input stream.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract FSDataInputStream open(Path gvfsPath, int bufferSize) throws
IOException;
+
+ /**
+ * Set the working directory. Same as {@link
FileSystem#setWorkingDirectory(Path)}.
+ *
+ * @param gvfsDir the new working directory.
+ * @throws FilesetPathNotFoundException if the fileset path is not found.
+ */
+ public abstract void setWorkingDirectory(Path gvfsDir) throws
FilesetPathNotFoundException;
+
+ /**
+ * Create a file. Same as {@link FileSystem#create(Path, FsPermission,
boolean, int, short, long,
+ * Progressable)}.
+ *
+ * @param gvfsPath the virtual path of the file.
+ * @param permission the permission.
+ * @param overwrite whether to overwrite the file if it exists.
+ * @param bufferSize the buffer size.
+ * @param replication the replication factor.
+ * @param blockSize the block size.
+ * @param progress the progressable.
+ * @return the output stream.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract FSDataOutputStream create(
+ Path gvfsPath,
+ FsPermission permission,
+ boolean overwrite,
+ int bufferSize,
+ short replication,
+ long blockSize,
+ Progressable progress)
+ throws IOException;
+
+ /**
+ * Append to an existing file. Same as {@link FileSystem#append(Path, int,
Progressable)}.
+ *
+ * @param gvfsPath the virtual path of the file.
+ * @param bufferSize the buffer size.
+ * @param progress the progressable.
+ * @return the output stream.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract FSDataOutputStream append(Path gvfsPath, int bufferSize,
Progressable progress)
+ throws IOException;
+
+ /**
+ * Rename a file. Same as {@link FileSystem#rename(Path, Path)}.
+ *
+ * @param srcGvfsPath the virtual source path.
+ * @param dstGvfsPath the virtual destination path.
+ * @return true if the rename is successful.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract boolean rename(Path srcGvfsPath, Path dstGvfsPath) throws
IOException;
+
+ /**
+ * Delete a file or directory. Same as {@link FileSystem#delete(Path,
boolean)}.
+ *
+ * @param gvfsPath the virtual path of the file or directory.
+ * @param recursive whether to delete the directory recursively.
+ * @return true if the deletion is successful.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract boolean delete(Path gvfsPath, boolean recursive) throws
IOException;
+
+ /**
+ * Get the file status. Same as {@link FileSystem#getFileStatus(Path)}.
+ *
+ * @param gvfsPath the virtual path of the file.
+ * @return the file status.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract FileStatus getFileStatus(Path gvfsPath) throws IOException;
+
+ /**
+ * List the statuses of the files/directories in the given path. Same as
{@link
+ * FileSystem#listStatus(Path)}.
+ *
+ * @param gvfsPath the virtual path of the directory.
+ * @return the array of file statuses.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract FileStatus[] listStatus(Path gvfsPath) throws IOException;
+
+ /**
+ * Make the given file and all non-existent parents directories. Same as
{@link
+ * FileSystem#mkdirs(Path, FsPermission)}.
+ *
+ * @param gvfsPath the virtual path of the directory.
+ * @param permission the permission.
+ * @return true if the directory creation is successful.
+ * @throws IOException if an I/O error occurs.
+ */
+ public abstract boolean mkdirs(Path gvfsPath, FsPermission permission)
throws IOException;
+
+ /**
+ * Get the default replication factor for the file. Same as {@link
+ * FileSystem#getDefaultReplication(Path)}.
+ *
+ * @param gvfsPath the virtual path of the file.
+ * @return the default replication factor.
+ */
+ public abstract short getDefaultReplication(Path gvfsPath);
+
+ /**
+ * Get the default block size for the file. Same as {@link
FileSystem#getDefaultBlockSize(Path)}.
+ *
+ * @param gvfsPath the virtual path of the file.
+ * @return the default block size.
+ */
+ public abstract long getDefaultBlockSize(Path gvfsPath);
+
+ /**
+ * Add delegation tokens to the credentials. Same as {@link
FileSystem#addDelegationTokens(String,
+ * Credentials)}.
+ *
+ * @param renewer the renewer.
+ * @param credentials the credentials.
+ * @return the array of tokens.
+ */
+ public abstract Token<?>[] addDelegationTokens(String renewer, Credentials
credentials);
+
+ /**
+ * Add delegation tokens for all file systems in the cache.
+ *
+ * @param renewer the renewer.
+ * @param credentials the credentials.
+ * @return the array of tokens.
+ */
+ protected Token<?>[] addDelegationTokensForAllFS(String renewer, Credentials
credentials) {
+ List<Token<?>> tokenList = Lists.newArrayList();
+ for (FileSystem fileSystem : internalFileSystemCache.asMap().values()) {
+ try {
+ tokenList.addAll(Arrays.asList(fileSystem.addDelegationTokens(renewer,
credentials)));
+ } catch (IOException e) {
+ LOG.warn("Failed to add delegation tokens for filesystem: {}",
fileSystem.getUri(), e);
+ }
+ }
+ return tokenList.stream().distinct().toArray(Token[]::new);
+ }
+
+ /**
+ * Get the current location name. The subclass can override this method to
provide a custom
+ * location name.
+ *
+ * @return the current location name, or null if you want to use the default
location.
+ */
+ @Nullable
+ protected String currentLocationName() {
+ return currentLocationName;
+ }
+
+ /**
+ * Get the metalake name.
+ *
+ * @return the metalake name.
+ */
+ protected String metalakeName() {
+ return metalakeName;
+ }
+
+ /**
+ * Get the default block size.
+ *
+ * @return the default block size.
+ */
+ protected long defaultBlockSize() {
+ return defaultBlockSize;
+ }
+
+ /**
+ * Get the actual file path by the given virtual path and location name.
+ *
+ * @param gvfsPath the virtual path.
+ * @param locationName the location name.
+ * @param operation the fileset data operation.
+ * @return the actual file path.
+ * @throws FilesetPathNotFoundException if the fileset path is not found.
+ */
+ protected Path getActualFilePath(
+ Path gvfsPath, String locationName, FilesetDataOperation operation)
+ throws FilesetPathNotFoundException {
+ NameIdentifier filesetIdent = extractIdentifier(metalakeName,
gvfsPath.toString());
+ String subPath = getSubPathFromGvfsPath(filesetIdent, gvfsPath.toString());
+ NameIdentifier catalogIdent =
+ NameIdentifier.of(filesetIdent.namespace().level(0),
filesetIdent.namespace().level(1));
+ setCallerContext(operation);
+ String fileLocation;
+ try {
+ fileLocation =
+ getFilesetCatalog(catalogIdent)
+ .getFileLocation(
+ NameIdentifier.of(filesetIdent.namespace().level(2),
filesetIdent.name()),
+ subPath,
+ locationName);
+ } catch (NoSuchCatalogException | CatalogNotInUseException e) {
+ String message = String.format("Cannot get fileset catalog by
identifier: %s", catalogIdent);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+
+ } catch (NoSuchFilesetException e) {
+ String message =
+ String.format(
+ "Cannot get fileset by fileset identifier: %s, sub_path %s",
filesetIdent, subPath);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+
+ } catch (NoSuchLocationNameException e) {
+ String message =
+ String.format(
+ "Location name not found by fileset identifier: %s, sub_path %s,
location_name %s",
+ filesetIdent, subPath, locationName);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+ }
+
+ Path actualFilePath = new Path(fileLocation);
+ URI uri = actualFilePath.toUri();
+ String scheme = uri.getScheme();
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(scheme), "Scheme of the actual file location
cannot be null.");
+ return new Path(fileLocation);
+ }
+
+ private void createFilesetLocationIfNeed(
+ NameIdentifier filesetIdent, FileSystem fs, Path filesetPath)
+ throws FilesetPathNotFoundException {
+ NameIdentifier catalogIdent =
+ NameIdentifier.of(filesetIdent.namespace().level(0),
filesetIdent.namespace().level(1));
+ // If the server-side filesystem ops are disabled, the fileset directory
may not exist. In such
+ // case the operations like create, open, list files under this directory
will fail. So we
+ // need to check the existence of the fileset directory beforehand.
+ boolean fsOpsDisabled =
+ ((Catalog) getFilesetCatalog(catalogIdent))
+ .properties()
+ .getOrDefault("disable-filesystem-ops", "false")
+ .equalsIgnoreCase("true");
+ if (fsOpsDisabled) {
+ try {
+ if (!fs.exists(filesetPath)) {
+ fs.mkdirs(filesetPath);
+ LOG.info(
+ "Automatically created a directory for fileset path: {} when "
+ + "disable-filesystem-ops sets to true in the catalog
properties.",
+ filesetPath);
+ }
+ } catch (IOException e) {
+ LOG.warn("Cannot create directory for fileset path: {}", filesetPath,
e);
+ }
+ }
+ }
+
+ /**
+ * Get the actual file system by the given virtual path.
+ *
+ * @param fileStatus the file status.
+ * @param actualPrefix the actual prefix.
+ * @param filesetPrefix the virtual prefix.
+ * @return the file status with the converted path.
+ */
+ protected FileStatus convertFileStatusPathPrefix(
+ FileStatus fileStatus, String actualPrefix, String filesetPrefix) {
+ String filePath = fileStatus.getPath().toString();
+ Preconditions.checkArgument(
+ filePath.startsWith(actualPrefix),
+ "Path %s doesn't start with prefix \"%s\".",
+ filePath,
+ actualPrefix);
+ // if the storage location ends with "/",
+ // we should truncate this to avoid replace issues.
+ Path path =
+ new Path(
+ filePath.replaceFirst(
+ actualPrefix.endsWith(SLASH) && !filesetPrefix.endsWith(SLASH)
+ ? actualPrefix.substring(0, actualPrefix.length() - 1)
+ : actualPrefix,
+ filesetPrefix));
+ fileStatus.setPath(path);
+
+ return fileStatus;
+ }
+
+ /**
+ * Get the virtual location by the given identifier.
+ *
+ * @param identifier the identifier.
+ * @param withScheme whether to include the scheme.
+ * @return the virtual location.
+ */
+ protected String getVirtualLocation(NameIdentifier identifier, boolean
withScheme) {
+ return String.format(
+ "%s/%s/%s/%s",
+ withScheme ?
GravitinoVirtualFileSystemConfiguration.GVFS_FILESET_PREFIX : "",
+ identifier.namespace().level(1),
+ identifier.namespace().level(2),
+ identifier.name());
+ }
+
+ /**
+ * Get the actual file system corresponding to the given virtual path and
location name.
+ *
+ * @param filesetPath the virtual path.
+ * @param locationName the location name. null means the default location.
+ * @return the actual file system.
+ * @throws FilesetPathNotFoundException if the fileset path is not found.
+ */
+ protected FileSystem getActualFileSystem(Path filesetPath, String
locationName)
+ throws FilesetPathNotFoundException {
+ NameIdentifier filesetIdent = extractIdentifier(metalakeName,
filesetPath.toString());
+ return getActualFileSystemByLocationName(filesetIdent, locationName);
+ }
+
+ /**
+ * Get the fileset catalog by the catalog identifier from the cache. If the
subclass does not want
+ * to use the cache, it can override this method.
+ *
+ * @param catalogIdent the catalog identifier.
+ * @return the fileset catalog.
+ */
+ protected FilesetCatalog getFilesetCatalog(NameIdentifier catalogIdent) {
+ return filesetCatalogCache.getFilesetCatalog(catalogIdent);
+ }
+
+ @VisibleForTesting
+ Cache<Pair<NameIdentifier, String>, FileSystem> internalFileSystemCache() {
+ return internalFileSystemCache;
+ }
+
+ private void setCallerContext(FilesetDataOperation operation) {
+ Map<String, String> contextMap = Maps.newHashMap();
+ contextMap.put(
+ FilesetAuditConstants.HTTP_HEADER_INTERNAL_CLIENT_TYPE,
+ InternalClientType.HADOOP_GVFS.name());
+ contextMap.put(FilesetAuditConstants.HTTP_HEADER_FILESET_DATA_OPERATION,
operation.name());
+ CallerContext callerContext =
CallerContext.builder().withContext(contextMap).build();
+ CallerContext.CallerContextHolder.set(callerContext);
+ }
+
+ private FileSystem getActualFileSystemByLocationName(
+ NameIdentifier filesetIdent, String locationName) throws
FilesetPathNotFoundException {
+ NameIdentifier catalogIdent =
+ NameIdentifier.of(filesetIdent.namespace().level(0),
filesetIdent.namespace().level(1));
+ try {
+ return internalFileSystemCache.get(
+ Pair.of(filesetIdent, locationName),
+ cacheKey -> {
+ try {
+ Fileset fileset = getFileset(cacheKey.getLeft());
+ String targetLocationName =
+ cacheKey.getRight() == null
+ ?
fileset.properties().get(PROPERTY_DEFAULT_LOCATION_NAME)
+ : cacheKey.getRight();
+
+ Preconditions.checkArgument(
+ fileset.storageLocations().containsKey(targetLocationName),
+ "Location name: %s is not found in fileset: %s.",
+ targetLocationName,
+ cacheKey.getLeft());
+
+ Path targetLocation = new
Path(fileset.storageLocations().get(targetLocationName));
+ Map<String, String> allProperties =
+ getAllProperties(cacheKey.getLeft(),
targetLocation.toUri().getScheme());
+
+ FileSystem actualFileSystem =
+ getActualFileSystemByPath(targetLocation, allProperties);
+ createFilesetLocationIfNeed(cacheKey.getLeft(),
actualFileSystem, targetLocation);
+ return actualFileSystem;
+ } catch (IOException ioe) {
+ throw new GravitinoRuntimeException(
+ ioe,
+ "Exception occurs when create new FileSystem for fileset:
%s, location: %s, msg: %s",
+ cacheKey.getLeft(),
+ cacheKey.getRight(),
+ ioe.getMessage());
+ }
+ });
+ } catch (RuntimeException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof NoSuchCatalogException || cause instanceof
CatalogNotInUseException) {
+ String message =
+ String.format("Cannot get fileset catalog by identifier: %s",
catalogIdent);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+ }
+
+ if (cause instanceof NoSuchFilesetException) {
+ String message =
+ String.format("Cannot get fileset by fileset identifier: %s",
filesetIdent);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+ }
+
+ if (cause instanceof NoSuchLocationNameException) {
+ String message =
+ String.format(
+ "Location name not found by fileset identifier: %s,
location_name %s",
+ filesetIdent, locationName);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+ }
+
+ throw e;
+ }
+ }
+
+ private FileSystem getActualFileSystemByPath(
+ Path actualFilePath, Map<String, String> allProperties) throws
IOException {
+ URI uri = actualFilePath.toUri();
+ String scheme = uri.getScheme();
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(scheme), "Scheme of the actual file location
cannot be null.");
+
+ FileSystemProvider provider = getFileSystemProviderByScheme(scheme);
+
+ // Reset the FileSystem service loader to make sure the FileSystem will
reload the
+ // service file systems, this is a temporary solution to fix the issue
+ // https://github.com/apache/gravitino/issues/5609
+ resetFileSystemServiceLoader(scheme);
+
+ return provider.getFileSystem(actualFilePath, allProperties);
+ }
+
+ private void resetFileSystemServiceLoader(String fsScheme) {
+ try {
+ Map<String, Class<? extends FileSystem>> serviceFileSystems =
+ (Map<String, Class<? extends FileSystem>>)
+ FieldUtils.getField(FileSystem.class, "SERVICE_FILE_SYSTEMS",
true).get(null);
+
+ if (serviceFileSystems.containsKey(fsScheme)) {
+ return;
+ }
+
+ // Set this value to false so that FileSystem will reload the service
file systems when
+ // needed.
+ FieldUtils.getField(FileSystem.class, "FILE_SYSTEMS_LOADED",
true).set(null, false);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Fileset getFileset(NameIdentifier filesetIdent) {
+ NameIdentifier catalogIdent =
+ NameIdentifier.of(filesetIdent.namespace().level(0),
filesetIdent.namespace().level(1));
+ return getFilesetCatalog(catalogIdent)
+ .loadFileset(NameIdentifier.of(filesetIdent.namespace().level(2),
filesetIdent.name()));
+ }
+
+ private Cache<Pair<NameIdentifier, String>, FileSystem> newFileSystemCache(
+ Configuration configuration, ScheduledThreadPoolExecutor
internalFileSystemCleanScheduler) {
+ int maxCapacity =
+ configuration.getInt(
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_FILESET_CACHE_MAX_CAPACITY_KEY,
+ GravitinoVirtualFileSystemConfiguration
+ .FS_GRAVITINO_FILESET_CACHE_MAX_CAPACITY_DEFAULT);
+ Preconditions.checkArgument(
+ maxCapacity > 0,
+ "'%s' should be greater than 0",
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_FILESET_CACHE_MAX_CAPACITY_KEY);
+
+ long evictionMillsAfterAccess =
+ configuration.getLong(
+ GravitinoVirtualFileSystemConfiguration
+ .FS_GRAVITINO_FILESET_CACHE_EVICTION_MILLS_AFTER_ACCESS_KEY,
+ GravitinoVirtualFileSystemConfiguration
+
.FS_GRAVITINO_FILESET_CACHE_EVICTION_MILLS_AFTER_ACCESS_DEFAULT);
+ Preconditions.checkArgument(
+ evictionMillsAfterAccess > 0,
+ "'%s' should be greater than 0",
+ GravitinoVirtualFileSystemConfiguration
+ .FS_GRAVITINO_FILESET_CACHE_EVICTION_MILLS_AFTER_ACCESS_KEY);
+
+ Caffeine<Object, Object> cacheBuilder =
+ Caffeine.newBuilder()
+ .maximumSize(maxCapacity)
+
.scheduler(Scheduler.forScheduledExecutorService(internalFileSystemCleanScheduler))
+ .removalListener(
+ (key, value, cause) -> {
+ FileSystem fs = (FileSystem) value;
+ if (fs != null) {
+ try {
+ fs.close();
+ } catch (IOException e) {
+ LOG.error("Cannot close the file system for fileset:
{}", key, e);
+ }
+ }
+ });
+ cacheBuilder.expireAfterAccess(evictionMillsAfterAccess,
TimeUnit.MILLISECONDS);
+ return cacheBuilder.build();
+ }
+
+ private Map<String, String> getAllProperties(NameIdentifier filesetIdent,
String scheme) {
+ Catalog catalog =
+ (Catalog)
+ getFilesetCatalog(
+ NameIdentifier.of(
+ filesetIdent.namespace().level(0),
filesetIdent.namespace().level(1)));
+
+ Map<String, String> allProperties =
getNecessaryProperties(catalog.properties());
+ allProperties.putAll(getConfigMap(conf));
+ allProperties.putAll(
+ getCredentialProperties(getFileSystemProviderByScheme(scheme),
filesetIdent));
+ return allProperties;
+ }
+
+ private Map<String, String> getNecessaryProperties(Map<String, String>
properties) {
+ return properties.entrySet().stream()
+ .filter(property ->
CATALOG_NECESSARY_PROPERTIES_TO_KEEP.contains(property.getKey()))
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+ }
+
+ private Map<String, String> getCredentialProperties(
+ FileSystemProvider fileSystemProvider, NameIdentifier filesetIdentifier)
{
+ // Do not support credential vending, we do not need to add any credential
properties.
+ if (!(fileSystemProvider instanceof SupportsCredentialVending)) {
+ return ImmutableMap.of();
+ }
+
+ ImmutableMap.Builder<String, String> mapBuilder = ImmutableMap.builder();
+ try {
+ Fileset fileset = getFileset(filesetIdentifier);
+ Credential[] credentials =
fileset.supportsCredentials().getCredentials();
+ if (credentials.length > 0) {
+ mapBuilder.put(
+ GravitinoFileSystemCredentialsProvider.GVFS_CREDENTIAL_PROVIDER,
+
DefaultGravitinoFileSystemCredentialsProvider.class.getCanonicalName());
+ mapBuilder.put(
+ GravitinoFileSystemCredentialsProvider.GVFS_NAME_IDENTIFIER,
+ filesetIdentifier.toString());
+
+ SupportsCredentialVending supportsCredentialVending =
+ (SupportsCredentialVending) fileSystemProvider;
+
mapBuilder.putAll(supportsCredentialVending.getFileSystemCredentialConf(credentials));
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ return mapBuilder.build();
+ }
+
+ private FileSystemProvider getFileSystemProviderByScheme(String scheme) {
+ FileSystemProvider provider = fileSystemProvidersMap.get(scheme);
+ if (provider == null) {
+ throw new GravitinoRuntimeException(
+ "Unsupported file system scheme: %s for %s.",
+ scheme, GravitinoVirtualFileSystemConfiguration.GVFS_SCHEME);
+ }
+ return provider;
+ }
+
+ private ThreadFactory newDaemonThreadFactory(String name) {
+ return new ThreadFactoryBuilder().setDaemon(true).setNameFormat(name +
"-%d").build();
+ }
+
+ private Map<String, FileSystemProvider> getFileSystemProviders() {
+ Map<String, FileSystemProvider> resultMap = Maps.newHashMap();
+ ServiceLoader<FileSystemProvider> allFileSystemProviders =
+ ServiceLoader.load(FileSystemProvider.class);
+
+ Streams.stream(allFileSystemProviders.iterator())
+ .forEach(
+ fileSystemProvider -> {
+ if (resultMap.containsKey(fileSystemProvider.scheme())) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "File system provider: '%s' with scheme '%s' already
exists in the provider list, "
+ + "please make sure the file system provider
scheme is unique.",
+ fileSystemProvider.getClass().getName(),
fileSystemProvider.scheme()));
+ }
+ resultMap.put(fileSystemProvider.scheme(), fileSystemProvider);
+ });
+ return resultMap;
+ }
+
+ private String initCurrentLocationName(Configuration configuration) {
+ // get from configuration first, otherwise use the env variable
+ // if both are not set, return null which means use the default location
+ return
Optional.ofNullable(configuration.get(FS_GRAVITINO_CURRENT_LOCATION_NAME))
+ .orElse(System.getenv(currentLocationEnvVar));
+ }
+}
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/DefaultGVFSOperations.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/DefaultGVFSOperations.java
new file mode 100644
index 0000000000..7390a7d26c
--- /dev/null
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/DefaultGVFSOperations.java
@@ -0,0 +1,234 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.filesystem.hadoop;
+
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.extractIdentifier;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.getSubPathFromGvfsPath;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.Arrays;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.audit.FilesetDataOperation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.Progressable;
+
+/**
+ * The default implementation of {@link BaseGVFSOperations} which takes
operations on the default
+ * location of the fileset if the location name is not provided in both
configuration and the
+ * environment.
+ */
+public class DefaultGVFSOperations extends BaseGVFSOperations {
+
+ /**
+ * Constructs a new {@link DefaultGVFSOperations} with the given {@link
Configuration}.
+ *
+ * @param configuration the configuration
+ */
+ public DefaultGVFSOperations(Configuration configuration) {
+ super(configuration);
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ }
+
+ @Override
+ public FSDataInputStream open(Path gvfsPath, int bufferSize) throws
IOException {
+ FileSystem actualFs = getActualFileSystem(gvfsPath, currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.OPEN);
+ return actualFs.open(actualFilePath, bufferSize);
+ }
+
+ @Override
+ public synchronized void setWorkingDirectory(Path gvfsDir) throws
FilesetPathNotFoundException {
+ FileSystem actualFs = getActualFileSystem(gvfsDir, currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsDir, currentLocationName(),
FilesetDataOperation.SET_WORKING_DIR);
+ actualFs.setWorkingDirectory(actualFilePath);
+ }
+
+ @Override
+ public FSDataOutputStream create(
+ Path gvfsPath,
+ FsPermission permission,
+ boolean overwrite,
+ int bufferSize,
+ short replication,
+ long blockSize,
+ Progressable progress)
+ throws IOException {
+ try {
+ FileSystem actualFs = getActualFileSystem(gvfsPath,
currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.CREATE);
+ return actualFs.create(
+ actualFilePath, permission, overwrite, bufferSize, replication,
blockSize, progress);
+ } catch (FilesetPathNotFoundException e) {
+ String message =
+ "Fileset is not found for path: "
+ + gvfsPath
+ + " for operation CREATE. "
+ + "This may be caused by fileset related metadata not found or
not in use in "
+ + "Gravitino, please check the fileset metadata in Gravitino.";
+ throw new IOException(message, e);
+ }
+ }
+
+ @Override
+ public FSDataOutputStream append(Path gvfsPath, int bufferSize, Progressable
progress)
+ throws IOException {
+ FileSystem actualFs = getActualFileSystem(gvfsPath, currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.APPEND);
+ return actualFs.append(actualFilePath, bufferSize, progress);
+ }
+
+ @Override
+ public boolean rename(Path srcGvfsPath, Path dstGvfsPath) throws IOException
{
+ // Fileset identifier is not allowed to be renamed, only its
subdirectories can be renamed
+ // which not in the storage location of the fileset;
+ NameIdentifier srcIdentifier = extractIdentifier(metalakeName(),
srcGvfsPath.toString());
+ NameIdentifier dstIdentifier = extractIdentifier(metalakeName(),
dstGvfsPath.toString());
+ Preconditions.checkArgument(
+ srcIdentifier.equals(dstIdentifier),
+ "Destination path fileset identifier: %s should be same with src path "
+ + "fileset identifier: %s.",
+ srcIdentifier,
+ dstIdentifier);
+
+ Path srcActualPath =
+ getActualFilePath(srcGvfsPath, currentLocationName(),
FilesetDataOperation.RENAME);
+ Path dstActualPath =
+ getActualFilePath(dstGvfsPath, currentLocationName(),
FilesetDataOperation.RENAME);
+ FileSystem actualFs = getActualFileSystem(srcGvfsPath,
currentLocationName());
+ return actualFs.rename(srcActualPath, dstActualPath);
+ }
+
+ @Override
+ public boolean delete(Path gvfsPath, boolean recursive) throws IOException {
+ try {
+ FileSystem actualFs = getActualFileSystem(gvfsPath,
currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.DELETE);
+ return actualFs.delete(actualFilePath, recursive);
+ } catch (FilesetPathNotFoundException e) {
+ return false;
+ }
+ }
+
+ @Override
+ public FileStatus getFileStatus(Path gvfsPath) throws IOException {
+ FileSystem actualFs = getActualFileSystem(gvfsPath, currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.GET_FILE_STATUS);
+ FileStatus fileStatus = actualFs.getFileStatus(actualFilePath);
+
+ NameIdentifier identifier = extractIdentifier(metalakeName(),
gvfsPath.toString());
+ String subPath = getSubPathFromGvfsPath(identifier, gvfsPath.toString());
+ String filesetLocation =
+ actualFilePath
+ .toString()
+ .substring(0, actualFilePath.toString().length() -
subPath.length());
+
+ return convertFileStatusPathPrefix(
+ fileStatus, filesetLocation, getVirtualLocation(identifier, true));
+ }
+
+ @Override
+ public FileStatus[] listStatus(Path gvfsPath) throws IOException {
+ FileSystem actualFs = getActualFileSystem(gvfsPath, currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.LIST_STATUS);
+ FileStatus[] fileStatusResults = actualFs.listStatus(actualFilePath);
+
+ NameIdentifier identifier = extractIdentifier(metalakeName(),
gvfsPath.toString());
+ String subPath = getSubPathFromGvfsPath(identifier, gvfsPath.toString());
+ String filesetLocation =
+ actualFilePath
+ .toString()
+ .substring(0, actualFilePath.toString().length() -
subPath.length());
+
+ return Arrays.stream(fileStatusResults)
+ .map(
+ fileStatus ->
+ convertFileStatusPathPrefix(
+ fileStatus, filesetLocation,
getVirtualLocation(identifier, true)))
+ .toArray(FileStatus[]::new);
+ }
+
+ @Override
+ public boolean mkdirs(Path gvfsPath, FsPermission permission) throws
IOException {
+ try {
+ FileSystem actualFs = getActualFileSystem(gvfsPath,
currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(gvfsPath, currentLocationName(),
FilesetDataOperation.MKDIRS);
+ return actualFs.mkdirs(actualFilePath, permission);
+ } catch (FilesetPathNotFoundException e) {
+ String message =
+ "Fileset is not found for path: "
+ + gvfsPath
+ + " for operation MKDIRS. "
+ + "This may be caused by fileset related metadata not found or
not in use in "
+ + "Gravitino, please check the fileset metadata in Gravitino.";
+ throw new IOException(message, e);
+ }
+ }
+
+ @Override
+ public short getDefaultReplication(Path gvfsPath) {
+ try {
+ FileSystem actualFs = getActualFileSystem(gvfsPath,
currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(
+ gvfsPath, currentLocationName(),
FilesetDataOperation.GET_DEFAULT_REPLICATION);
+ return actualFs.getDefaultReplication(actualFilePath);
+ } catch (FilesetPathNotFoundException e) {
+ return 1;
+ }
+ }
+
+ @Override
+ public long getDefaultBlockSize(Path gvfsPath) {
+ try {
+ FileSystem actualFs = getActualFileSystem(gvfsPath,
currentLocationName());
+ Path actualFilePath =
+ getActualFilePath(
+ gvfsPath, currentLocationName(),
FilesetDataOperation.GET_DEFAULT_BLOCK_SIZE);
+ return actualFs.getDefaultBlockSize(actualFilePath);
+ } catch (FilesetPathNotFoundException e) {
+ return defaultBlockSize();
+ }
+ }
+
+ @Override
+ public Token<?>[] addDelegationTokens(String renewer, Credentials
credentials) {
+ return addDelegationTokensForAllFS(renewer, credentials);
+ }
+}
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetCatalogCache.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetCatalogCache.java
new file mode 100644
index 0000000000..33e40ef20e
--- /dev/null
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetCatalogCache.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.filesystem.hadoop;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.Scheduler;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.client.GravitinoClient;
+import org.apache.gravitino.file.FilesetCatalog;
+
+/** A cache for fileset catalogs. */
+public class FilesetCatalogCache implements Closeable {
+
+ private final GravitinoClient client;
+ private final Cache<NameIdentifier, FilesetCatalog> catalogCache;
+
+ // Since Caffeine does not ensure that removalListener will be involved
after expiration
+ // We use a scheduler with one thread to clean up expired clients.
+ private final ScheduledThreadPoolExecutor catalogCleanScheduler;
+
+ /**
+ * Creates a new instance of {@link FilesetCatalogCache}.
+ *
+ * @param client the Gravitino client.
+ */
+ public FilesetCatalogCache(GravitinoClient client) {
+ this.client = client;
+ this.catalogCleanScheduler =
+ new ScheduledThreadPoolExecutor(1,
newDaemonThreadFactory("gvfs-catalog-cache-cleaner"));
+ this.catalogCache = newCatalogCache(catalogCleanScheduler);
+ }
+
+ /**
+ * Gets the fileset catalog by the given catalog identifier.
+ *
+ * @param catalogIdent the catalog identifier.
+ * @return the fileset catalog.
+ */
+ public FilesetCatalog getFilesetCatalog(NameIdentifier catalogIdent) {
+ FilesetCatalog filesetCatalog =
+ catalogCache.get(
+ catalogIdent, ident ->
client.loadCatalog(catalogIdent.name()).asFilesetCatalog());
+
+ Preconditions.checkArgument(
+ filesetCatalog != null, String.format("Loaded fileset catalog: %s is
null.", catalogIdent));
+ return filesetCatalog;
+ }
+
+ private Cache<NameIdentifier, FilesetCatalog> newCatalogCache(
+ ScheduledThreadPoolExecutor catalogCleanScheduler) {
+ // In most scenarios, it will not read so many catalog filesets at the
same time, so we can just
+ // set a default value for this cache.
+ return Caffeine.newBuilder()
+ .maximumSize(100)
+
.scheduler(Scheduler.forScheduledExecutorService(catalogCleanScheduler))
+ .build();
+ }
+
+ private ThreadFactory newDaemonThreadFactory(String name) {
+ return new ThreadFactoryBuilder().setDaemon(true).setNameFormat(name +
"-%d").build();
+ }
+
+ @Override
+ public void close() throws IOException {
+ catalogCache.invalidateAll();
+ // close the client
+ try {
+ if (client != null) {
+ client.close();
+ }
+ } catch (Exception e) {
+ // ignore
+ }
+ catalogCleanScheduler.shutdownNow();
+ }
+}
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetPathNotFoundException.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetPathNotFoundException.java
index f9f26808fb..c0fbc6760f 100644
---
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetPathNotFoundException.java
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/FilesetPathNotFoundException.java
@@ -36,4 +36,15 @@ public class FilesetPathNotFoundException extends
FileNotFoundException {
public FilesetPathNotFoundException(String message) {
super(message);
}
+
+ /**
+ * Creates a new FilesetPathNotFoundException instance with the given
message and cause.
+ *
+ * @param message The message of the exception.
+ * @param cause The cause of the exception.
+ */
+ public FilesetPathNotFoundException(String message, Throwable cause) {
+ super(message);
+ initCause(cause);
+ }
}
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystem.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystem.java
index dd2763933e..673f3e50d2 100644
---
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystem.java
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystem.java
@@ -18,59 +18,16 @@
*/
package org.apache.gravitino.filesystem.hadoop;
-import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CURRENT_LOCATION_NAME;
-import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.extractIdentifier;
-import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.getConfigMap;
-import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.getSubPathFromGvfsPath;
-
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.Scheduler;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Supplier;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import com.google.common.collect.Streams;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
import java.net.URI;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.ServiceLoader;
-import java.util.Set;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import javax.annotation.Nullable;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.reflect.FieldUtils;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.gravitino.Catalog;
-import org.apache.gravitino.NameIdentifier;
-import org.apache.gravitino.audit.CallerContext;
-import org.apache.gravitino.audit.FilesetAuditConstants;
import org.apache.gravitino.audit.FilesetDataOperation;
-import org.apache.gravitino.audit.InternalClientType;
-import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
-import
org.apache.gravitino.catalog.hadoop.fs.GravitinoFileSystemCredentialsProvider;
-import org.apache.gravitino.catalog.hadoop.fs.SupportsCredentialVending;
-import org.apache.gravitino.client.GravitinoClient;
-import org.apache.gravitino.credential.Credential;
import org.apache.gravitino.exceptions.CatalogNotInUseException;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchFilesetException;
-import org.apache.gravitino.file.Fileset;
-import org.apache.gravitino.file.FilesetCatalog;
-import org.apache.gravitino.storage.AzureProperties;
-import org.apache.gravitino.storage.OSSProperties;
-import org.apache.gravitino.storage.S3Properties;
+import org.apache.gravitino.exceptions.NoSuchLocationNameException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -95,29 +52,7 @@ public class GravitinoVirtualFileSystem extends FileSystem {
private Path workingDirectory;
private URI uri;
- private GravitinoClient client;
- private String metalakeName;
- private Cache<NameIdentifier, FilesetCatalog> catalogCache;
- private ScheduledThreadPoolExecutor catalogCleanScheduler;
- // Fileset nameIdentifier-locationName Pair and its corresponding FileSystem
cache, the name
- // identifier has four levels, the first level is metalake name.
- private Cache<Pair<NameIdentifier, String>, FileSystem>
internalFileSystemCache;
- private ScheduledThreadPoolExecutor internalFileSystemCleanScheduler;
- private long defaultBlockSize;
-
- private static final String SLASH = "/";
- private final Map<String, FileSystemProvider> fileSystemProvidersMap =
Maps.newHashMap();
- private String currentLocationEnvVar;
-
- @Nullable private String currentLocationName;
-
- private static final Set<String> CATALOG_NECESSARY_PROPERTIES_TO_KEEP =
- Sets.newHashSet(
- OSSProperties.GRAVITINO_OSS_ENDPOINT,
- OSSProperties.GRAVITINO_OSS_REGION,
- S3Properties.GRAVITINO_S3_ENDPOINT,
- S3Properties.GRAVITINO_S3_REGION,
- AzureProperties.GRAVITINO_AZURE_STORAGE_ACCOUNT_NAME);
+ private BaseGVFSOperations operations;
@Override
public void initialize(URI name, Configuration configuration) throws
IOException {
@@ -128,86 +63,34 @@ public class GravitinoVirtualFileSystem extends FileSystem
{
name.getScheme(),
GravitinoVirtualFileSystemConfiguration.GVFS_SCHEME));
}
- int maxCapacity =
- configuration.getInt(
-
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_FILESET_CACHE_MAX_CAPACITY_KEY,
- GravitinoVirtualFileSystemConfiguration
- .FS_GRAVITINO_FILESET_CACHE_MAX_CAPACITY_DEFAULT);
- Preconditions.checkArgument(
- maxCapacity > 0,
- "'%s' should be greater than 0",
-
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_FILESET_CACHE_MAX_CAPACITY_KEY);
-
- long evictionMillsAfterAccess =
- configuration.getLong(
- GravitinoVirtualFileSystemConfiguration
- .FS_GRAVITINO_FILESET_CACHE_EVICTION_MILLS_AFTER_ACCESS_KEY,
- GravitinoVirtualFileSystemConfiguration
-
.FS_GRAVITINO_FILESET_CACHE_EVICTION_MILLS_AFTER_ACCESS_DEFAULT);
- Preconditions.checkArgument(
- evictionMillsAfterAccess > 0,
- "'%s' should be greater than 0",
- GravitinoVirtualFileSystemConfiguration
- .FS_GRAVITINO_FILESET_CACHE_EVICTION_MILLS_AFTER_ACCESS_KEY);
-
- initializeFileSystemCache(maxCapacity, evictionMillsAfterAccess);
- initializeCatalogCache();
-
- this.metalakeName =
-
configuration.get(GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_METALAKE_KEY);
- Preconditions.checkArgument(
- StringUtils.isNotBlank(metalakeName),
- "'%s' is not set in the configuration",
-
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CLIENT_METALAKE_KEY);
-
- this.client = GravitinoVirtualFileSystemUtils.createClient(configuration);
- // Register the default local and HDFS FileSystemProvider
- fileSystemProvidersMap.putAll(getFileSystemProviders());
-
- this.currentLocationEnvVar =
+ String operationsClassName =
configuration.get(
-
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_CURRENT_LOCATION_NAME_ENV_VAR,
- GravitinoVirtualFileSystemConfiguration
- .FS_GRAVITINO_CURRENT_LOCATION_NAME_ENV_VAR_DEFAULT);
- this.currentLocationName = initCurrentLocationName(configuration);
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_OPERATIONS_CLASS,
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_OPERATIONS_CLASS_DEFAULT);
+ try {
+ Class<? extends BaseGVFSOperations> operationsClz =
+ (Class<? extends BaseGVFSOperations>)
Class.forName(operationsClassName);
+ this.operations =
+
operationsClz.getDeclaredConstructor(Configuration.class).newInstance(configuration);
+ } catch (Exception e) {
+ if (e instanceof InvocationTargetException
+ && ((InvocationTargetException) e).getTargetException() instanceof
RuntimeException) {
+ throw (RuntimeException) ((InvocationTargetException)
e).getTargetException();
+ }
+ throw new GravitinoRuntimeException(
+ e, "Cannot create operations instance: %s", operationsClassName);
+ }
this.workingDirectory = new Path(name);
this.uri = URI.create(name.getScheme() + "://" + name.getAuthority());
- this.defaultBlockSize =
- configuration.getLong(
- GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE,
-
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE_DEFAULT);
setConf(configuration);
super.initialize(uri, getConf());
}
@VisibleForTesting
- Cache<Pair<NameIdentifier, String>, FileSystem> internalFileSystemCache() {
- return internalFileSystemCache;
- }
-
- @VisibleForTesting
- FileStatus convertFileStatusPathPrefix(
- FileStatus fileStatus, String actualPrefix, String virtualPrefix) {
- String filePath = fileStatus.getPath().toString();
- Preconditions.checkArgument(
- filePath.startsWith(actualPrefix),
- "Path %s doesn't start with prefix \"%s\".",
- filePath,
- actualPrefix);
- // if the storage location ends with "/",
- // we should truncate this to avoid replace issues.
- Path path =
- new Path(
- filePath.replaceFirst(
- actualPrefix.endsWith(SLASH) && !virtualPrefix.endsWith(SLASH)
- ? actualPrefix.substring(0, actualPrefix.length() - 1)
- : actualPrefix,
- virtualPrefix));
- fileStatus.setPath(path);
-
- return fileStatus;
+ BaseGVFSOperations getOperations() {
+ return operations;
}
@Override
@@ -222,25 +105,23 @@ public class GravitinoVirtualFileSystem extends
FileSystem {
@Override
public synchronized void setWorkingDirectory(Path newDir) {
- Optional<FilesetContextPair> context =
- getFilesetContext(newDir, FilesetDataOperation.SET_WORKING_DIR);
try {
- throwFilesetPathNotFoundExceptionIf(
- () -> !context.isPresent(), newDir,
FilesetDataOperation.SET_WORKING_DIR);
+ runWithExceptionTranslation(
+ () -> {
+ operations.setWorkingDirectory(newDir);
+ return null;
+ },
+ FilesetDataOperation.SET_WORKING_DIR);
} catch (FilesetPathNotFoundException e) {
throw new RuntimeException(e);
}
-
-
context.get().getFileSystem().setWorkingDirectory(context.get().getActualFileLocation());
this.workingDirectory = newDir;
}
@Override
public FSDataInputStream open(Path path, int bufferSize) throws IOException {
- Optional<FilesetContextPair> context = getFilesetContext(path,
FilesetDataOperation.OPEN);
- throwFilesetPathNotFoundExceptionIf(
- () -> !context.isPresent(), path, FilesetDataOperation.OPEN);
- return
context.get().getFileSystem().open(context.get().getActualFileLocation(),
bufferSize);
+ return runWithExceptionTranslation(
+ () -> operations.open(path, bufferSize), FilesetDataOperation.OPEN);
}
@Override
@@ -253,445 +134,144 @@ public class GravitinoVirtualFileSystem extends
FileSystem {
long blockSize,
Progressable progress)
throws IOException {
- Optional<FilesetContextPair> context = getFilesetContext(path,
FilesetDataOperation.CREATE);
- if (!context.isPresent()) {
- throw new IOException(
+ try {
+ return operations.create(
+ path, permission, overwrite, bufferSize, replication, blockSize,
progress);
+ } catch (NoSuchCatalogException
+ | CatalogNotInUseException
+ | NoSuchFilesetException
+ | NoSuchLocationNameException
+ | FilesetPathNotFoundException e) {
+ String message =
"Fileset is not found for path: "
+ path
+ " for operation CREATE. "
+ "This may be caused by fileset related metadata not found or
not in use in "
- + "Gravitino, please check the fileset metadata in Gravitino.");
+ + "Gravitino, please check the fileset metadata in Gravitino.";
+ throw new IOException(message, e);
}
-
- return context
- .get()
- .getFileSystem()
- .create(
- context.get().getActualFileLocation(),
- permission,
- overwrite,
- bufferSize,
- replication,
- blockSize,
- progress);
}
@Override
public FSDataOutputStream append(Path path, int bufferSize, Progressable
progress)
throws IOException {
- Optional<FilesetContextPair> context = getFilesetContext(path,
FilesetDataOperation.APPEND);
- throwFilesetPathNotFoundExceptionIf(
- () -> !context.isPresent(), path, FilesetDataOperation.APPEND);
- return context
- .get()
- .getFileSystem()
- .append(context.get().getActualFileLocation(), bufferSize, progress);
+ return runWithExceptionTranslation(
+ () -> operations.append(path, bufferSize, progress),
FilesetDataOperation.APPEND);
}
@Override
public boolean rename(Path src, Path dst) throws IOException {
- // Fileset identifier is not allowed to be renamed, only its
subdirectories can be renamed
- // which not in the storage location of the fileset;
- NameIdentifier srcIdentifier = extractIdentifier(metalakeName,
src.toString());
- NameIdentifier dstIdentifier = extractIdentifier(metalakeName,
dst.toString());
- Preconditions.checkArgument(
- srcIdentifier.equals(dstIdentifier),
- "Destination path fileset identifier: %s should be same with src path "
- + "fileset identifier: %s.",
- srcIdentifier,
- dstIdentifier);
-
- Optional<FilesetContextPair> srcContext = getFilesetContext(src,
FilesetDataOperation.RENAME);
- throwFilesetPathNotFoundExceptionIf(
- () -> !srcContext.isPresent(), src, FilesetDataOperation.RENAME);
-
- Optional<FilesetContextPair> dstContext = getFilesetContext(dst,
FilesetDataOperation.RENAME);
-
- // Because src context and dst context are the same, so if src context is
present, dst context
- // must be present.
- return srcContext
- .get()
- .getFileSystem()
- .rename(srcContext.get().getActualFileLocation(),
dstContext.get().getActualFileLocation());
+ return runWithExceptionTranslation(
+ () -> operations.rename(src, dst), FilesetDataOperation.RENAME);
}
@Override
public boolean delete(Path path, boolean recursive) throws IOException {
- Optional<FilesetContextPair> context = getFilesetContext(path,
FilesetDataOperation.DELETE);
- if (context.isPresent()) {
- return
context.get().getFileSystem().delete(context.get().getActualFileLocation(),
recursive);
- } else {
+ try {
+ return runWithExceptionTranslation(
+ () -> operations.delete(path, recursive),
FilesetDataOperation.DELETE);
+ } catch (FilesetPathNotFoundException e) {
return false;
}
}
@Override
public FileStatus getFileStatus(Path path) throws IOException {
- Optional<FilesetContextPair> context =
- getFilesetContext(path, FilesetDataOperation.GET_FILE_STATUS);
- throwFilesetPathNotFoundExceptionIf(
- () -> !context.isPresent(), path,
FilesetDataOperation.GET_FILE_STATUS);
-
- FileStatus fileStatus =
-
context.get().getFileSystem().getFileStatus(context.get().getActualFileLocation());
- NameIdentifier identifier = extractIdentifier(metalakeName,
path.toString());
- String subPath = getSubPathFromGvfsPath(identifier, path.toString());
- String storageLocation =
- context
- .get()
- .getActualFileLocation()
- .toString()
- .substring(
- 0, context.get().getActualFileLocation().toString().length() -
subPath.length());
- return convertFileStatusPathPrefix(
- fileStatus, storageLocation, getVirtualLocation(identifier, true));
+ return runWithExceptionTranslation(
+ () -> operations.getFileStatus(path),
FilesetDataOperation.GET_FILE_STATUS);
}
@Override
public FileStatus[] listStatus(Path path) throws IOException {
- Optional<FilesetContextPair> context =
- getFilesetContext(path, FilesetDataOperation.LIST_STATUS);
- throwFilesetPathNotFoundExceptionIf(
- () -> !context.isPresent(), path, FilesetDataOperation.LIST_STATUS);
-
- FileStatus[] fileStatusResults =
-
context.get().getFileSystem().listStatus(context.get().getActualFileLocation());
- NameIdentifier identifier = extractIdentifier(metalakeName,
path.toString());
- String subPath = getSubPathFromGvfsPath(identifier, path.toString());
- String storageLocation =
- context
- .get()
- .getActualFileLocation()
- .toString()
- .substring(
- 0, context.get().getActualFileLocation().toString().length() -
subPath.length());
- return Arrays.stream(fileStatusResults)
- .map(
- fileStatus ->
- convertFileStatusPathPrefix(
- fileStatus, storageLocation,
getVirtualLocation(identifier, true)))
- .toArray(FileStatus[]::new);
+ return runWithExceptionTranslation(
+ () -> operations.listStatus(path), FilesetDataOperation.LIST_STATUS);
}
@Override
public boolean mkdirs(Path path, FsPermission permission) throws IOException
{
- Optional<FilesetContextPair> context = getFilesetContext(path,
FilesetDataOperation.MKDIRS);
- if (!context.isPresent()) {
- throw new IOException(
+ try {
+ return operations.mkdirs(path, permission);
+ } catch (NoSuchCatalogException
+ | CatalogNotInUseException
+ | NoSuchFilesetException
+ | NoSuchLocationNameException
+ | FilesetPathNotFoundException e) {
+ String message =
"Fileset is not found for path: "
+ path
+ " for operation MKDIRS. "
+ "This may be caused by fileset related metadata not found or
not in use in "
- + "Gravitino, please check the fileset metadata in Gravitino.");
+ + "Gravitino, please check the fileset metadata in Gravitino.";
+ throw new IOException(message, e);
}
-
- return
context.get().getFileSystem().mkdirs(context.get().getActualFileLocation(),
permission);
}
@Override
public short getDefaultReplication(Path f) {
- Optional<FilesetContextPair> context =
- getFilesetContext(f, FilesetDataOperation.GET_DEFAULT_REPLICATION);
- return context
- .map(c ->
c.getFileSystem().getDefaultReplication(c.getActualFileLocation()))
- .orElse((short) 1);
+ try {
+ return runWithExceptionTranslation(
+ () -> operations.getDefaultReplication(f),
FilesetDataOperation.GET_DEFAULT_REPLICATION);
+ } catch (IOException e) {
+ return 1;
+ }
}
@Override
public long getDefaultBlockSize(Path f) {
- Optional<FilesetContextPair> context =
- getFilesetContext(f, FilesetDataOperation.GET_DEFAULT_BLOCK_SIZE);
- return context
- .map(c ->
c.getFileSystem().getDefaultBlockSize(c.getActualFileLocation()))
- .orElse(defaultBlockSize);
+ try {
+ return runWithExceptionTranslation(
+ () -> operations.getDefaultBlockSize(f),
FilesetDataOperation.GET_DEFAULT_BLOCK_SIZE);
+ } catch (IOException e) {
+ return operations.defaultBlockSize();
+ }
}
@Override
public Token<?>[] addDelegationTokens(String renewer, Credentials
credentials) {
- List<Token<?>> tokenList = Lists.newArrayList();
- for (FileSystem fileSystem : internalFileSystemCache.asMap().values()) {
- try {
- tokenList.addAll(Arrays.asList(fileSystem.addDelegationTokens(renewer,
credentials)));
- } catch (IOException e) {
- LOG.warn("Failed to add delegation tokens for filesystem: {}",
fileSystem.getUri(), e);
- }
- }
- return tokenList.stream().distinct().toArray(Token[]::new);
+ return operations.addDelegationTokens(renewer, credentials);
}
@Override
public synchronized void close() throws IOException {
- // close all actual FileSystems
- for (FileSystem fileSystem : internalFileSystemCache.asMap().values()) {
- try {
- fileSystem.close();
- } catch (IOException e) {
- // ignore
- }
- }
- internalFileSystemCache.invalidateAll();
- catalogCache.invalidateAll();
- // close the client
try {
- if (client != null) {
- client.close();
- }
- } catch (Exception e) {
- // ignore
+ operations.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close operations: {}",
operations.getClass().getName(), e);
}
- catalogCleanScheduler.shutdownNow();
- internalFileSystemCleanScheduler.shutdownNow();
- super.close();
- }
- private void initializeFileSystemCache(int maxCapacity, long
expireAfterAccess) {
- // Since Caffeine does not ensure that removalListener will be involved
after expiration
- // We use a scheduler with one thread to clean up expired clients.
- this.internalFileSystemCleanScheduler =
- new ScheduledThreadPoolExecutor(1,
newDaemonThreadFactory("gvfs-filesystem-cache-cleaner"));
- Caffeine<Object, Object> cacheBuilder =
- Caffeine.newBuilder()
- .maximumSize(maxCapacity)
-
.scheduler(Scheduler.forScheduledExecutorService(internalFileSystemCleanScheduler))
- .removalListener(
- (key, value, cause) -> {
- FileSystem fs = (FileSystem) value;
- if (fs != null) {
- try {
- fs.close();
- } catch (IOException e) {
- LOG.error("Cannot close the file system for fileset:
{}", key, e);
- }
- }
- });
- if (expireAfterAccess > 0) {
- cacheBuilder.expireAfterAccess(expireAfterAccess, TimeUnit.MILLISECONDS);
- }
- this.internalFileSystemCache = cacheBuilder.build();
- }
-
- private void initializeCatalogCache() {
- // Since Caffeine does not ensure that removalListener will be involved
after expiration
- // We use a scheduler with one thread to clean up expired clients.
- this.catalogCleanScheduler =
- new ScheduledThreadPoolExecutor(1,
newDaemonThreadFactory("gvfs-catalog-cache-cleaner"));
- // In most scenarios, it will not read so many catalog filesets at the
same time, so we can just
- // set a default value for this cache.
- this.catalogCache =
- Caffeine.newBuilder()
- .maximumSize(100)
-
.scheduler(Scheduler.forScheduledExecutorService(catalogCleanScheduler))
- .build();
- }
-
- private String initCurrentLocationName(Configuration configuration) {
- // get from configuration first, otherwise use the env variable
- // if both are not set, return null which means use the default location
- return
Optional.ofNullable(configuration.get(FS_GRAVITINO_CURRENT_LOCATION_NAME))
- .orElse(System.getenv(currentLocationEnvVar));
- }
-
- private ThreadFactory newDaemonThreadFactory(String name) {
- return new ThreadFactoryBuilder().setDaemon(true).setNameFormat(name +
"-%d").build();
- }
-
- private String getVirtualLocation(NameIdentifier identifier, boolean
withScheme) {
- return String.format(
- "%s/%s/%s/%s",
- withScheme ?
GravitinoVirtualFileSystemConfiguration.GVFS_FILESET_PREFIX : "",
- identifier.namespace().level(1),
- identifier.namespace().level(2),
- identifier.name());
+ super.close();
}
- private Optional<FilesetContextPair> getFilesetContext(
- Path virtualPath, FilesetDataOperation operation) {
- NameIdentifier identifier = extractIdentifier(metalakeName,
virtualPath.toString());
- String virtualPathString = virtualPath.toString();
- String subPath = getSubPathFromGvfsPath(identifier, virtualPathString);
- NameIdentifier catalogIdent = NameIdentifier.of(metalakeName,
identifier.namespace().level(1));
-
- FilesetCatalog filesetCatalog;
+ private <R, E extends IOException> R runWithExceptionTranslation(
+ Executable<R, E> executable, FilesetDataOperation operation)
+ throws FilesetPathNotFoundException, E {
try {
- filesetCatalog =
- catalogCache.get(
- catalogIdent, ident ->
client.loadCatalog(catalogIdent.name()).asFilesetCatalog());
- Preconditions.checkArgument(
- filesetCatalog != null,
- String.format("Loaded fileset catalog: %s is null.", catalogIdent));
+ return executable.execute();
} catch (NoSuchCatalogException | CatalogNotInUseException e) {
- LOG.warn("Cannot get fileset catalog by identifier: {}", catalogIdent,
e);
- return Optional.empty();
- }
+ String message = String.format("Cannot get fileset catalog during %s",
operation);
+ LOG.warn(message, operation, e);
+ throw new FilesetPathNotFoundException(message, e);
- Map<String, String> contextMap = Maps.newHashMap();
- contextMap.put(
- FilesetAuditConstants.HTTP_HEADER_INTERNAL_CLIENT_TYPE,
- InternalClientType.HADOOP_GVFS.name());
- contextMap.put(FilesetAuditConstants.HTTP_HEADER_FILESET_DATA_OPERATION,
operation.name());
- CallerContext callerContext =
CallerContext.builder().withContext(contextMap).build();
- CallerContext.CallerContextHolder.set(callerContext);
-
- String actualFileLocation;
- try {
- actualFileLocation =
- filesetCatalog.getFileLocation(
- NameIdentifier.of(identifier.namespace().level(2),
identifier.name()),
- subPath,
- currentLocationName);
} catch (NoSuchFilesetException e) {
- LOG.warn("Cannot get file location by identifier: {}, sub_path {}",
identifier, subPath, e);
- return Optional.empty();
- }
-
- Path filePath = new Path(actualFileLocation);
- URI uri = filePath.toUri();
- // we cache the fs for the same scheme, so we can reuse it
- String scheme = uri.getScheme();
- Preconditions.checkArgument(
- StringUtils.isNotBlank(scheme), "Scheme of the actual file location
cannot be null.");
- FileSystem fs =
- internalFileSystemCache.get(
- Pair.of(identifier, currentLocationName),
- ident -> {
- try {
- FileSystemProvider provider =
fileSystemProvidersMap.get(scheme);
- if (provider == null) {
- throw new GravitinoRuntimeException(
- "Unsupported file system scheme: %s for %s.",
- scheme,
GravitinoVirtualFileSystemConfiguration.GVFS_SCHEME);
- }
-
- // Reset the FileSystem service loader to make sure the
FileSystem will reload the
- // service file systems, this is a temporary solution to fix
the issue
- // https://github.com/apache/gravitino/issues/5609
- resetFileSystemServiceLoader(scheme);
-
- Catalog catalog = (Catalog) filesetCatalog;
- Map<String, String> necessaryPropertyFromCatalog =
- catalog.properties().entrySet().stream()
- .filter(
- property ->
-
CATALOG_NECESSARY_PROPERTIES_TO_KEEP.contains(property.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
-
- Map<String, String> totalProperty =
Maps.newHashMap(necessaryPropertyFromCatalog);
- totalProperty.putAll(getConfigMap(getConf()));
- totalProperty.putAll(getCredentialProperties(provider,
catalog, identifier));
- return provider.getFileSystem(filePath, totalProperty);
-
- } catch (IOException ioe) {
- throw new GravitinoRuntimeException(
- ioe,
- "Exception occurs when create new FileSystem for actual
uri: %s, msg: %s",
- uri,
- ioe.getMessage());
- }
- });
-
- return Optional.of(new FilesetContextPair(new Path(actualFileLocation),
fs));
- }
-
- private Map<String, String> getCredentialProperties(
- FileSystemProvider fileSystemProvider, Catalog catalog, NameIdentifier
filesetIdentifier) {
- // Do not support credential vending, we do not need to add any credential
properties.
- if (!(fileSystemProvider instanceof SupportsCredentialVending)) {
- return ImmutableMap.of();
- }
-
- ImmutableMap.Builder<String, String> mapBuilder = ImmutableMap.builder();
- try {
- Fileset fileset =
- catalog
- .asFilesetCatalog()
- .loadFileset(
- NameIdentifier.of(
- filesetIdentifier.namespace().level(2),
filesetIdentifier.name()));
- Credential[] credentials =
fileset.supportsCredentials().getCredentials();
- if (credentials.length > 0) {
- mapBuilder.put(
- GravitinoFileSystemCredentialsProvider.GVFS_CREDENTIAL_PROVIDER,
-
DefaultGravitinoFileSystemCredentialsProvider.class.getCanonicalName());
- mapBuilder.put(
- GravitinoFileSystemCredentialsProvider.GVFS_NAME_IDENTIFIER,
- filesetIdentifier.toString());
-
- SupportsCredentialVending supportsCredentialVending =
- (SupportsCredentialVending) fileSystemProvider;
-
mapBuilder.putAll(supportsCredentialVending.getFileSystemCredentialConf(credentials));
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- return mapBuilder.build();
- }
-
- private void resetFileSystemServiceLoader(String fsScheme) {
- try {
- Map<String, Class<? extends FileSystem>> serviceFileSystems =
- (Map<String, Class<? extends FileSystem>>)
- FieldUtils.getField(FileSystem.class, "SERVICE_FILE_SYSTEMS",
true).get(null);
-
- if (serviceFileSystems.containsKey(fsScheme)) {
- return;
+ String message = String.format("Cannot get fileset during %s",
operation);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+
+ } catch (NoSuchLocationNameException e) {
+ String message = String.format("Cannot find location name during %s",
operation);
+ LOG.warn(message, e);
+ throw new FilesetPathNotFoundException(message, e);
+
+ } catch (IOException e) {
+ if (e instanceof FilesetPathNotFoundException) {
+ throw (FilesetPathNotFoundException) e;
}
-
- // Set this value to false so that FileSystem will reload the service
file systems when
- // needed.
- FieldUtils.getField(FileSystem.class, "FILE_SYSTEMS_LOADED",
true).set(null, false);
- } catch (Exception e) {
- throw new RuntimeException(e);
+ throw e;
}
}
- private static Map<String, FileSystemProvider> getFileSystemProviders() {
- Map<String, FileSystemProvider> resultMap = Maps.newHashMap();
- ServiceLoader<FileSystemProvider> allFileSystemProviders =
- ServiceLoader.load(FileSystemProvider.class);
-
- Streams.stream(allFileSystemProviders.iterator())
- .forEach(
- fileSystemProvider -> {
- if (resultMap.containsKey(fileSystemProvider.scheme())) {
- throw new UnsupportedOperationException(
- String.format(
- "File system provider: '%s' with scheme '%s' already
exists in the provider list, "
- + "please make sure the file system provider
scheme is unique.",
- fileSystemProvider.getClass().getName(),
fileSystemProvider.scheme()));
- }
- resultMap.put(fileSystemProvider.scheme(), fileSystemProvider);
- });
- return resultMap;
- }
-
- private void throwFilesetPathNotFoundExceptionIf(
- Supplier<Boolean> condition, Path path, FilesetDataOperation op)
- throws FilesetPathNotFoundException {
- if (condition.get()) {
- throw new FilesetPathNotFoundException(
- String.format(
- "Path [%s] not found for operation [%s] because of fileset and
related "
- + "metadata not existed in Gravitino",
- path, op));
- }
- }
-
- private static class FilesetContextPair {
- private final Path actualFileLocation;
- private final FileSystem fileSystem;
-
- public FilesetContextPair(Path actualFileLocation, FileSystem fileSystem) {
- this.actualFileLocation = actualFileLocation;
- this.fileSystem = fileSystem;
- }
-
- public Path getActualFileLocation() {
- return actualFileLocation;
- }
-
- public FileSystem getFileSystem() {
- return fileSystem;
- }
+ @FunctionalInterface
+ private interface Executable<R, E extends Exception> {
+ R execute() throws E;
}
}
diff --git
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
index 658eb617f6..c73b2cb477 100644
---
a/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
+++
b/clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java
@@ -116,6 +116,13 @@ public class GravitinoVirtualFileSystemConfiguration {
public static final String
FS_GRAVITINO_CURRENT_LOCATION_NAME_ENV_VAR_DEFAULT =
"CURRENT_LOCATION_NAME";
+ /** The configuration key for the GVFS operations class. */
+ public static final String FS_GRAVITINO_OPERATIONS_CLASS =
"fs.gravitino.operations.class";
+
+ /** The default value for the GVFS operations class. */
+ public static final String FS_GRAVITINO_OPERATIONS_CLASS_DEFAULT =
+ DefaultGVFSOperations.class.getCanonicalName();
+
/** The configuration key for the block size of the GVFS file. */
public static final String FS_GRAVITINO_BLOCK_SIZE =
"fs.gravitino.block.size";
diff --git
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
index 0a4e1cd908..ba5ca5a161 100644
---
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
+++
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/GravitinoMockServerBase.java
@@ -18,7 +18,6 @@
*/
package org.apache.gravitino.filesystem.hadoop;
-import static org.apache.gravitino.file.Fileset.LOCATION_NAME_UNKNOWN;
import static org.apache.hc.core5.http.HttpStatus.SC_OK;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -180,28 +179,25 @@ public abstract class GravitinoMockServerBase {
String schemaName,
String filesetName,
Fileset.Type type,
- String location) {
+ Map<String, String> locations,
+ Map<String, String> properties) {
NameIdentifier fileset = NameIdentifier.of(metalakeName, catalogName,
schemaName, filesetName);
- String filesetPath =
+ String filesetEndpoint =
String.format(
"/api/metalakes/%s/catalogs/%s/schemas/%s/filesets/%s",
metalakeName, catalogName, schemaName,
RESTUtils.encodeString(filesetName));
- Map<String, String> locations =
- location == null
- ? Collections.emptyMap()
- : ImmutableMap.of(LOCATION_NAME_UNKNOWN, location);
FilesetDTO mockFileset =
FilesetDTO.builder()
.name(fileset.name())
.type(type)
.storageLocations(locations)
.comment("comment")
- .properties(ImmutableMap.of("k1", "v1"))
+ .properties(properties)
.audit(AuditDTO.builder().withCreator("creator").withCreateTime(Instant.now()).build())
.build();
FilesetResponse filesetResponse = new FilesetResponse(mockFileset);
try {
- buildMockResource(Method.GET, filesetPath, null, filesetResponse, SC_OK);
+ buildMockResource(Method.GET, filesetEndpoint, null, filesetResponse,
SC_OK);
} catch (JsonProcessingException e) {
throw new RuntimeException(e);
}
diff --git
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
index 31d7d078cf..62ee35f12c 100644
---
a/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
+++
b/clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/TestGvfsBase.java
@@ -19,6 +19,8 @@
package org.apache.gravitino.filesystem.hadoop;
import static org.apache.gravitino.file.Fileset.LOCATION_NAME_UNKNOWN;
+import static org.apache.gravitino.file.Fileset.PROPERTY_DEFAULT_LOCATION_NAME;
+import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE_DEFAULT;
import static
org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemUtils.extractIdentifier;
import static org.apache.hc.core5.http.HttpStatus.SC_NOT_FOUND;
import static org.apache.hc.core5.http.HttpStatus.SC_OK;
@@ -28,10 +30,16 @@ import static
org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyShort;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.google.common.collect.ImmutableMap;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -51,7 +59,9 @@ import org.apache.gravitino.dto.responses.CredentialResponse;
import org.apache.gravitino.dto.responses.ErrorResponse;
import org.apache.gravitino.dto.responses.FileLocationResponse;
import org.apache.gravitino.dto.responses.FilesetResponse;
+import org.apache.gravitino.exceptions.NoSuchCatalogException;
import org.apache.gravitino.exceptions.NoSuchFilesetException;
+import org.apache.gravitino.exceptions.NoSuchLocationNameException;
import org.apache.gravitino.file.Fileset;
import org.apache.gravitino.rest.RESTUtils;
import org.apache.hadoop.conf.Configuration;
@@ -61,6 +71,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hc.core5.http.Method;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
@@ -68,6 +79,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
public class TestGvfsBase extends GravitinoMockServerBase {
protected static final String GVFS_IMPL_CLASS =
GravitinoVirtualFileSystem.class.getName();
@@ -111,6 +123,115 @@ public class TestGvfsBase extends GravitinoMockServerBase
{
mockCatalogDTO(catalogName, provider, "comment");
}
+ @Test
+ public void testOpsException() throws IOException, NoSuchFieldException,
IllegalAccessException {
+ Assumptions.assumeTrue(getClass() == TestGvfsBase.class);
+ Configuration newConf = new Configuration(conf);
+ try (GravitinoVirtualFileSystem fs =
+ (GravitinoVirtualFileSystem) new
Path("gvfs://fileset/").getFileSystem(newConf)) {
+ BaseGVFSOperations mockOps = Mockito.mock(BaseGVFSOperations.class);
+ // inject the mockOps
+ Field operationsField =
GravitinoVirtualFileSystem.class.getDeclaredField("operations");
+ operationsField.setAccessible(true);
+ operationsField.set(fs, mockOps);
+
+ // test setWorkingDirectory
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .setWorkingDirectory(any());
+ assertThrows(
+ RuntimeException.class, () -> fs.setWorkingDirectory(new
Path("gvfs://fileset/")));
+
+ // test open
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .open(any(), anyInt());
+ assertThrows(
+ FilesetPathNotFoundException.class, () -> fs.open(new
Path("gvfs://fileset/"), 1024));
+
+ // test create
+ Mockito.doThrow(new NoSuchCatalogException("fileset catalog not found"))
+ .when(mockOps)
+ .create(any(), any(), anyBoolean(), anyInt(), anyShort(), anyLong(),
any());
+ Exception exception =
+ assertThrows(IOException.class, () -> fs.create(new
Path("gvfs://fileset/"), true));
+ assertTrue(
+ exception.getMessage().contains("please check the fileset metadata
in Gravitino"),
+ "The expected message is: " + exception.getMessage());
+
+ // test append
+ Mockito.doThrow(new NoSuchLocationNameException("location name not
found"))
+ .when(mockOps)
+ .append(any(), anyInt(), any());
+ assertThrows(
+ FilesetPathNotFoundException.class, () -> fs.append(new
Path("gvfs://fileset/"), 1024));
+
+ // test rename
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .rename(any(), any());
+ assertThrows(
+ FilesetPathNotFoundException.class,
+ () -> fs.rename(new Path("gvfs://fileset/"), new
Path("gvfs://fileset/new")));
+
+ // test delete
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .delete(any(), anyBoolean());
+ assertEquals(false, fs.delete(new Path("gvfs://fileset/"), true));
+
+ // test getFileStatus
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .getFileStatus(any());
+ assertThrows(
+ FilesetPathNotFoundException.class, () -> fs.getFileStatus(new
Path("gvfs://fileset/")));
+
+ // test listStatus
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .listStatus(any());
+ assertThrows(
+ FilesetPathNotFoundException.class, () -> fs.listStatus(new
Path("gvfs://fileset/")));
+
+ // test listStatus
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .listStatus(any());
+ assertThrows(
+ FilesetPathNotFoundException.class, () -> fs.listStatus(new
Path("gvfs://fileset/")));
+
+ // test mkdirs
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .mkdirs(any(), any());
+ exception = assertThrows(IOException.class, () -> fs.mkdirs(new
Path("gvfs://fileset/")));
+ assertTrue(
+ exception.getMessage().contains("please check the fileset metadata
in Gravitino"),
+ "The expected message is: " + exception.getMessage());
+
+ // test getDefaultReplication
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .getDefaultReplication(any());
+ assertEquals(1, fs.getDefaultReplication(new Path("gvfs://fileset/")));
+
+ // test getDefaultBlockSize
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .getDefaultBlockSize(any());
+
Mockito.doReturn(FS_GRAVITINO_BLOCK_SIZE_DEFAULT).when(mockOps).defaultBlockSize();
+ assertEquals(
+ FS_GRAVITINO_BLOCK_SIZE_DEFAULT, fs.getDefaultBlockSize(new
Path("gvfs://fileset/")));
+
+ // test addDelegationTokens
+ Mockito.doThrow(new NoSuchFilesetException("fileset not found"))
+ .when(mockOps)
+ .addDelegationTokens(any(), any());
+ assertThrows(NoSuchFilesetException.class, () ->
fs.addDelegationTokens("renewer", null));
+ }
+ }
+
@Test
public void testFSCache() throws IOException {
String filesetName = "testFSCache";
@@ -158,6 +279,7 @@ public class TestGvfsBase extends GravitinoMockServerBase {
FileSystem proxyLocalFs =
Objects.requireNonNull(
((GravitinoVirtualFileSystem) gravitinoFileSystem)
+ .getOperations()
.internalFileSystemCache()
.getIfPresent(
Pair.of(
@@ -206,12 +328,17 @@ public class TestGvfsBase extends GravitinoMockServerBase
{
() ->
assertEquals(
0,
- ((GravitinoVirtualFileSystem)
fs).internalFileSystemCache().asMap().size()));
+ ((GravitinoVirtualFileSystem) fs)
+ .getOperations()
+ .internalFileSystemCache()
+ .asMap()
+ .size()));
assertNull(
((GravitinoVirtualFileSystem) fs)
+ .getOperations()
.internalFileSystemCache()
- .getIfPresent(Pair.of(NameIdentifier.of("file"), null)));
+ .getIfPresent(Pair.of(NameIdentifier.of("file"),
LOCATION_NAME_UNKNOWN)));
}
}
@@ -235,6 +362,14 @@ public class TestGvfsBase extends GravitinoMockServerBase {
FileSystemTestUtils.mkdirs(localPath, localFileSystem);
assertTrue(localFileSystem.exists(localPath));
// test gvfs normal create
+ mockFilesetDTO(
+ metalakeName,
+ catalogName,
+ schemaName,
+ filesetName,
+ Fileset.Type.MANAGED,
+ ImmutableMap.of("location1", localPath.toString()),
+ ImmutableMap.of(PROPERTY_DEFAULT_LOCATION_NAME, "location1"));
FileLocationResponse fileLocationResponse = new
FileLocationResponse(localPath + "/test.txt");
Map<String, String> queryParams = new HashMap<>();
queryParams.put("sub_path", RESTUtils.encodeString("/test.txt"));
@@ -323,32 +458,6 @@ public class TestGvfsBase extends GravitinoMockServerBase {
}
}
- private void buildMockResourceForCredential(String filesetName, String
filesetLocation)
- throws JsonProcessingException {
- String filesetPath =
- String.format(
- "/api/metalakes/%s/catalogs/%s/schemas/%s/filesets/%s",
- metalakeName, catalogName, schemaName, filesetName);
- String credentialsPath =
- String.format(
- "/api/metalakes/%s/objects/fileset/%s.%s.%s/credentials",
- metalakeName, catalogName, schemaName, filesetName);
- FilesetResponse filesetResponse =
- new FilesetResponse(
- FilesetDTO.builder()
- .name(filesetName)
- .comment("comment")
- .type(Fileset.Type.MANAGED)
- .audit(AuditDTO.builder().build())
- .storageLocations(ImmutableMap.of(LOCATION_NAME_UNKNOWN,
filesetLocation))
- .build());
- CredentialResponse credentialResponse = new CredentialResponse(new
CredentialDTO[] {});
-
- buildMockResource(Method.GET, filesetPath, ImmutableMap.of(), null,
filesetResponse, SC_OK);
- buildMockResource(
- Method.GET, credentialsPath, ImmutableMap.of(), null,
credentialResponse, SC_OK);
- }
-
@ParameterizedTest
@CsvSource({
"true, testRename",
@@ -743,7 +852,8 @@ public class TestGvfsBase extends GravitinoMockServerBase {
String storageLocation = "hdfs://hive:9000/";
String virtualLocation = "gvfs://fileset/test_catalog/tmp/test_fileset";
FileStatus convertedStatus =
- fs.convertFileStatusPathPrefix(fileStatus, storageLocation,
virtualLocation);
+ fs.getOperations()
+ .convertFileStatusPathPrefix(fileStatus, storageLocation,
virtualLocation);
Path expectedPath = new
Path("gvfs://fileset/test_catalog/tmp/test_fileset/test");
assertEquals(expectedPath, convertedStatus.getPath());
}
@@ -786,9 +896,34 @@ public class TestGvfsBase extends GravitinoMockServerBase {
assertThrows(IOException.class, () -> fs.mkdirs(testPath));
assertEquals(1, fs.getDefaultReplication(testPath));
- assertEquals(
-
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_BLOCK_SIZE_DEFAULT,
- fs.getDefaultBlockSize(testPath));
+ assertEquals(FS_GRAVITINO_BLOCK_SIZE_DEFAULT,
fs.getDefaultBlockSize(testPath));
}
}
+
+ private void buildMockResourceForCredential(String filesetName, String
filesetLocation)
+ throws JsonProcessingException {
+ String filesetPath =
+ String.format(
+ "/api/metalakes/%s/catalogs/%s/schemas/%s/filesets/%s",
+ metalakeName, catalogName, schemaName,
RESTUtils.encodeString(filesetName));
+ String credentialsPath =
+ String.format(
+ "/api/metalakes/%s/objects/fileset/%s.%s.%s/credentials",
+ metalakeName, catalogName, schemaName,
RESTUtils.encodeString(filesetName));
+ FilesetResponse filesetResponse =
+ new FilesetResponse(
+ FilesetDTO.builder()
+ .name(filesetName)
+ .comment("comment")
+ .type(Fileset.Type.MANAGED)
+ .audit(AuditDTO.builder().build())
+ .storageLocations(ImmutableMap.of(LOCATION_NAME_UNKNOWN,
filesetLocation))
+ .properties(ImmutableMap.of(PROPERTY_DEFAULT_LOCATION_NAME,
LOCATION_NAME_UNKNOWN))
+ .build());
+ CredentialResponse credentialResponse = new CredentialResponse(new
CredentialDTO[] {});
+
+ buildMockResource(Method.GET, filesetPath, ImmutableMap.of(), null,
filesetResponse, SC_OK);
+ buildMockResource(
+ Method.GET, credentialsPath, ImmutableMap.of(), null,
credentialResponse, SC_OK);
+ }
}
diff --git a/docs/how-to-use-gvfs.md b/docs/how-to-use-gvfs.md
index e31ab902a0..f03dc041a3 100644
--- a/docs/how-to-use-gvfs.md
+++ b/docs/how-to-use-gvfs.md
@@ -48,25 +48,26 @@ the path mapping and convert automatically.
### Configuration
-| Configuration item | Description
| Default value
| Required | Since version
|
-|-------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------------------------------------|-------------------------------------|------------------|
-| `fs.AbstractFileSystem.gvfs.impl` | The Gravitino
Virtual File System abstract class, set it to
`org.apache.gravitino.filesystem.hadoop.Gvfs`.
| (none) | Yes
| 0.5.0 |
-| `fs.gvfs.impl` | The Gravitino
Virtual File System implementation class, set it to
`org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystem`.
| (none) | Yes
| 0.5.0 |
-| `fs.gvfs.impl.disable.cache` | Disable the
Gravitino Virtual File System cache in the Hadoop environment. If you need to
proxy multi-user operations, please set this value to `true` and create a
separate File System for each user.
| `false`
| No
| 0.5.0 |
-| `fs.gravitino.server.uri` | The Gravitino server
URI which GVFS needs to load the fileset metadata.
| (none)
| Yes | 0.5.0
|
-| `fs.gravitino.client.metalake` | The metalake to
which the fileset belongs.
| (none)
| Yes | 0.5.0
|
-| `fs.gravitino.client.authType` | The auth type to
initialize the Gravitino client to use with the Gravitino Virtual File System.
Currently only supports `simple`, `oauth2` and `kerberos` auth types.
| `simple`
| No | 0.5.0
|
-| `fs.gravitino.client.oauth2.serverUri` | The auth server URI
for the Gravitino client when using `oauth2` auth type with the Gravitino
Virtual File System.
| (none)
| Yes if you use `oauth2` auth type | 0.5.0
|
-| `fs.gravitino.client.oauth2.credential` | The auth credential
for the Gravitino client when using `oauth2` auth type in the Gravitino Virtual
File System.
| (none)
| Yes if you use `oauth2` auth type | 0.5.0
|
-| `fs.gravitino.client.oauth2.path` | The auth server path
for the Gravitino client when using `oauth2` auth type with the Gravitino
Virtual File System. Please remove the first slash `/` from the path, for
example `oauth/token`.
| (none)
| Yes if you use `oauth2` auth type |
0.5.0 |
-| `fs.gravitino.client.oauth2.scope` | The auth scope for
the Gravitino client when using `oauth2` auth type with the Gravitino Virtual
File System.
| (none)
| Yes if you use `oauth2` auth type | 0.5.0
|
-| `fs.gravitino.client.kerberos.principal` | The auth principal
for the Gravitino client when using `kerberos` auth type with the Gravitino
Virtual File System.
| (none)
| Yes if you use `kerberos` auth type | 0.5.1
|
-| `fs.gravitino.client.kerberos.keytabFilePath` | The auth keytab file
path for the Gravitino client when using `kerberos` auth type in the Gravitino
Virtual File System.
| (none)
| No | 0.5.1
|
-| `fs.gravitino.fileset.cache.maxCapacity` | The cache capacity
of the Gravitino Virtual File System.
| `20`
| No | 0.5.0
|
-| `fs.gravitino.fileset.cache.evictionMillsAfterAccess` | The value of time
that the cache expires after accessing in the Gravitino Virtual File System.
The value is in `milliseconds`.
| `3600000`
| No | 0.5.0
|
-| `fs.gravitino.fileset.cache.evictionMillsAfterAccess` | The value of time
that the cache expires after accessing in the Gravitino Virtual File System.
The value is in `milliseconds`.
| `3600000`
| No | 0.5.0
|
-| `fs.gravitino.current.location.name` | The configuration
used to select the location of the fileset. If this configuration is not set,
the value of environment variable configured by
`fs.gravitino.current.location.env.var` will be checked. If neither is set, the
value of fileset property `default-location-name` will be used as the location
name. | the value of fileset property `default-location-name` | No
| 0.9.0-incubating |
-| `fs.gravitino.current.location.name.env.var` | The environment
variable name to get the current location name.
| `CURRENT_LOCATION_NAME`
| No |
0.9.0-incubating |
+| Configuration item | Description
| Default value
| Required | Since
version |
+|-------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------|-------------------------------------|------------------|
+| `fs.AbstractFileSystem.gvfs.impl` | The Gravitino
Virtual File System abstract class, set it to
`org.apache.gravitino.filesystem.hadoop.Gvfs`.
| (none) | Yes
| 0.5.0 |
+| `fs.gvfs.impl` | The Gravitino
Virtual File System implementation class, set it to
`org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystem`.
| (none) | Yes
| 0.5.0 |
+| `fs.gvfs.impl.disable.cache` | Disable the
Gravitino Virtual File System cache in the Hadoop environment. If you need to
proxy multi-user operations, please set this value to `true` and create a
separate File System for each user.
| `false`
| No
| 0.5.0 |
+| `fs.gravitino.server.uri` | The Gravitino server
URI which GVFS needs to load the fileset metadata.
| (none)
| Yes | 0.5.0
|
+| `fs.gravitino.client.metalake` | The metalake to
which the fileset belongs.
| (none)
| Yes |
0.5.0 |
+| `fs.gravitino.client.authType` | The auth type to
initialize the Gravitino client to use with the Gravitino Virtual File System.
Currently only supports `simple`, `oauth2` and `kerberos` auth types.
| `simple`
| No |
0.5.0 |
+| `fs.gravitino.client.oauth2.serverUri` | The auth server URI
for the Gravitino client when using `oauth2` auth type with the Gravitino
Virtual File System.
| (none)
| Yes if you use `oauth2` auth type |
0.5.0 |
+| `fs.gravitino.client.oauth2.credential` | The auth credential
for the Gravitino client when using `oauth2` auth type in the Gravitino Virtual
File System.
| (none)
| Yes if you use `oauth2` auth type | 0.5.0
|
+| `fs.gravitino.client.oauth2.path` | The auth server path
for the Gravitino client when using `oauth2` auth type with the Gravitino
Virtual File System. Please remove the first slash `/` from the path, for
example `oauth/token`.
| (none)
| Yes if you use `oauth2` auth
type | 0.5.0 |
+| `fs.gravitino.client.oauth2.scope` | The auth scope for
the Gravitino client when using `oauth2` auth type with the Gravitino Virtual
File System.
| (none)
| Yes if you use `oauth2` auth type |
0.5.0 |
+| `fs.gravitino.client.kerberos.principal` | The auth principal
for the Gravitino client when using `kerberos` auth type with the Gravitino
Virtual File System.
| (none)
| Yes if you use `kerberos` auth type |
0.5.1 |
+| `fs.gravitino.client.kerberos.keytabFilePath` | The auth keytab file
path for the Gravitino client when using `kerberos` auth type in the Gravitino
Virtual File System.
| (none)
| No | 0.5.1
|
+| `fs.gravitino.fileset.cache.maxCapacity` | The cache capacity
of the Gravitino Virtual File System.
| `20`
| No |
0.5.0 |
+| `fs.gravitino.fileset.cache.evictionMillsAfterAccess` | The value of time
that the cache expires after accessing in the Gravitino Virtual File System.
The value is in `milliseconds`.
| `3600000`
| No |
0.5.0 |
+| `fs.gravitino.fileset.cache.evictionMillsAfterAccess` | The value of time
that the cache expires after accessing in the Gravitino Virtual File System.
The value is in `milliseconds`.
| `3600000`
| No |
0.5.0 |
+| `fs.gravitino.current.location.name` | The configuration
used to select the location of the fileset. If this configuration is not set,
the value of environment variable configured by
`fs.gravitino.current.location.env.var` will be checked. If neither is set, the
value of fileset property `default-location-name` will be used as the location
name. | the value of fileset property `default-location-name` | No
| 0.9.0-incubating |
+| `fs.gravitino.current.location.name.env.var` | The environment
variable name to get the current location name.
| `CURRENT_LOCATION_NAME`
| No |
0.9.0-incubating |
+| `fs.gravitino.operations.class` | The operations class
to provide the FS operations for the Gravitino Virtual File System. Users can
extends `BaseGVFSOperations` to implement their own operations and configure
the class name in this conf to use custom FS operations.
|
`org.apache.gravitino.filesystem.hadoop.DefaultGVFSOperations` | No
| 0.9.0-incubating |
Apart from the above properties, to access fileset like S3, GCS, OSS and
custom fileset, extra properties are needed, please see
[S3 GVFS Java client
configurations](./hadoop-catalog-with-s3.md#using-the-gvfs-java-client-to-access-the-fileset),
[GCS GVFS Java client
configurations](./hadoop-catalog-with-gcs.md#using-the-gvfs-java-client-to-access-the-fileset),
[OSS GVFS Java client
configurations](./hadoop-catalog-with-oss.md#using-the-gvfs-java-client-to-access-the-fileset)
and [Azure Blob Storage GVFS Java client
configurations](./hadoop-catalog-with-adls.md#using-the-gvfs-java-client-to-access-the-fileset)
for [...]
