[#5582] improvement(hadoop3-filesystem): Remove configuration fs.gvfs.filesystem.providers from GVFS client.

build.gradle.kts

@@ -781,7 +781,7 @@ tasks {
         !it.name.startsWith("client") && !it.name.startsWith("filesystem") && !it.name.startsWith("spark") && !it.name.startsWith("iceberg") && it.name != "trino-connector" &&
         it.name != "integration-test" && it.name != "bundled-catalog" && !it.name.startsWith("flink") &&
         it.name != "integration-test" && it.name != "hive-metastore-common" && !it.name.startsWith("flink") &&
-        it.name != "gcp-bundle" && it.name != "aliyun-bundle" && it.name != "aws-bundle" && it.name != "azure-bundle"
+        it.name != "gcp-bundle" && it.name != "aliyun-bundle" && it.name != "aws-bundle" && it.name != "azure-bundle" && it.name != "hadoop-common"
       ) {
         from(it.configurations.runtimeClasspath)
         into("distribution/package/libs")
@@ -804,7 +804,7 @@ tasks {
         it.name != "bundled-catalog" &&
         it.name != "hive-metastore-common" && it.name != "gcp-bundle" &&
         it.name != "aliyun-bundle" && it.name != "aws-bundle" && it.name != "azure-bundle" &&
-        it.name != "docs"
+        it.name != "hadoop-common" && it.name != "docs"
       ) {
         dependsOn("${it.name}:build")
         from("${it.name}/build/libs")

bundles/aliyun-bundle/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
   compileOnly(libs.hadoop3.common)
 
   implementation(libs.aliyun.credentials.sdk)

bundles/aws-bundle/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
   compileOnly(libs.hadoop3.common)
 
   implementation(libs.aws.iam)

bundles/azure-bundle/build.gradle.kts

@@ -28,6 +28,9 @@ dependencies {
   compileOnly(project(":api"))
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
 
   compileOnly(libs.hadoop3.common)
 

bundles/gcp-bundle/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
   compileOnly(project(":core"))
   compileOnly(project(":catalogs:catalog-common"))
   compileOnly(project(":catalogs:catalog-hadoop"))
+  compileOnly(project(":catalogs:hadoop-common")) {
+    exclude("*")
+  }
 
   compileOnly(libs.hadoop3.common)
 

catalogs/catalog-hadoop/build.gradle.kts

@@ -32,10 +32,15 @@ dependencies {
   implementation(project(":core")) {
     exclude(group = "*")
   }
+
   implementation(project(":common")) {
     exclude(group = "*")
   }
 
+  implementation(project(":catalogs:hadoop-common")) {
+    exclude(group = "*")
+  }
+
   implementation(project(":catalogs:catalog-common")) {
     exclude(group = "*")
   }

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/fs/FileSystemUtils.java

@@ -63,8 +63,8 @@ public static Map<String, FileSystemProvider> getFileSystemProviders(String file
               if (resultMap.containsKey(fileSystemProvider.scheme())) {
                 throw new UnsupportedOperationException(
                     String.format(
-                        "File system provider: '%s' with scheme '%s' already exists in the use provider list "
-                            + "Please make sure the file system provider scheme is unique.",
+                        "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);

catalogs/hadoop-common/build.gradle.kts

@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+plugins {
+  id("java")
+}
+
+// try to avoid adding extra dependencies because it is used by catalogs and connectors.
+dependencies {
+  implementation(libs.commons.lang3)
+  implementation(libs.hadoop3.common)
+}

clients/filesystem-hadoop3/build.gradle.kts

@@ -26,7 +26,8 @@ plugins {
 dependencies {
   compileOnly(project(":clients:client-java-runtime", configuration = "shadow"))
   compileOnly(libs.hadoop3.common)
-  implementation(project(":catalogs:catalog-hadoop")) {
+
+  implementation(project(":catalogs:hadoop-common")) {
     exclude(group = "*")
   }
   implementation(project(":catalogs:catalog-common")) {
@@ -42,6 +43,7 @@ dependencies {
   testImplementation(project(":server-common"))
   testImplementation(project(":clients:client-java"))
   testImplementation(project(":integration-test-common", "testArtifacts"))
+  testImplementation(project(":catalogs:catalog-hadoop"))
   testImplementation(project(":bundles:gcp-bundle"))
   testImplementation(project(":bundles:aliyun-bundle"))
   testImplementation(project(":bundles:aws-bundle"))

clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystem.java

@@ -18,22 +18,22 @@
  */
 package org.apache.gravitino.filesystem.hadoop;
 
-import static org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_FILESYSTEM_PROVIDERS;
-
 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.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Streams;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.File;
 import java.io.IOException;
 import java.net.URI;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.ServiceLoader;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
@@ -46,7 +46,6 @@
 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.FileSystemUtils;
 import org.apache.gravitino.client.DefaultOAuth2TokenProvider;
 import org.apache.gravitino.client.GravitinoClient;
 import org.apache.gravitino.client.KerberosTokenProvider;
@@ -135,8 +134,7 @@ public void initialize(URI name, Configuration configuration) throws IOException
     initializeClient(configuration);
 
     // Register the default local and HDFS FileSystemProvider
-    String fileSystemProviders = configuration.get(FS_FILESYSTEM_PROVIDERS);
-    fileSystemProvidersMap.putAll(FileSystemUtils.getFileSystemProviders(fileSystemProviders));
+    fileSystemProvidersMap.putAll(getFileSystemProviders());
 
     this.workingDirectory = new Path(name);
     this.uri = URI.create(name.getScheme() + "://" + name.getAuthority());
@@ -618,4 +616,24 @@ public FileSystem getFileSystem() {
       return fileSystem;
     }
   }
+
+  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;
+  }
 }

clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/GravitinoVirtualFileSystemConfiguration.java

@@ -18,8 +18,6 @@
  */
 package org.apache.gravitino.filesystem.hadoop;
 
-import org.apache.gravitino.catalog.hadoop.fs.FileSystemProvider;
-
 /** Configuration class for Gravitino Virtual File System. */
 public class GravitinoVirtualFileSystemConfiguration {
 
@@ -44,13 +42,6 @@ public class GravitinoVirtualFileSystemConfiguration {
   /** The configuration key for the Gravitino client auth type. */
   public static final String FS_GRAVITINO_CLIENT_AUTH_TYPE_KEY = "fs.gravitino.client.authType";
 
-  /**
-   * File system provider names configuration key. The value is a comma separated list of file
-   * system provider name which is defined in the service loader. Users can custom their own file
-   * system by implementing the {@link FileSystemProvider} interface.
-   */
-  public static final String FS_FILESYSTEM_PROVIDERS = "fs.gvfs.filesystem.providers";
-
   /** The authentication type for simple authentication. */
   public static final String SIMPLE_AUTH_TYPE = "simple";
   /** The authentication type for oauth2 authentication. */

clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/integration/test/GravitinoVirtualFileSystemABSIT.java

@@ -95,7 +95,6 @@ public void startUp() throws Exception {
     conf.set("fs.gravitino.server.uri", serverUri);
     conf.set("fs.gravitino.client.metalake", metalakeName);
 
-    conf.set("fs.gvfs.filesystem.providers", AzureFileSystemProvider.ABS_PROVIDER_NAME);
     // Pass this configuration to the real file system
     conf.set(ABSProperties.GRAVITINO_ABS_ACCOUNT_NAME, ABS_ACCOUNT_NAME);
     conf.set(ABSProperties.GRAVITINO_ABS_ACCOUNT_KEY, ABS_ACCOUNT_KEY);

clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/integration/test/GravitinoVirtualFileSystemGCSIT.java

@@ -20,7 +20,6 @@
 package org.apache.gravitino.filesystem.hadoop.integration.test;
 
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS;
-import static org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_FILESYSTEM_PROVIDERS;
 
 import com.google.common.collect.Maps;
 import java.io.IOException;
@@ -92,15 +91,14 @@ public void startUp() throws Exception {
 
     // Pass this configuration to the real file system
     conf.set(GCSProperties.GCS_SERVICE_ACCOUNT_JSON_PATH, SERVICE_ACCOUNT_FILE);
-    conf.set(FS_FILESYSTEM_PROVIDERS, "gcs");
   }
 
   @AfterAll
   public void tearDown() throws IOException {
     Catalog catalog = metalake.loadCatalog(catalogName);
     catalog.asSchemas().dropSchema(schemaName, true);
-    metalake.dropCatalog(catalogName);
-    client.dropMetalake(metalakeName);
+    metalake.dropCatalog(catalogName, true);
+    client.dropMetalake(metalakeName, true);
 
     if (client != null) {
       client.close();

clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/integration/test/GravitinoVirtualFileSystemOSSIT.java

@@ -20,7 +20,6 @@
 package org.apache.gravitino.filesystem.hadoop.integration.test;
 
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS;
-import static org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_FILESYSTEM_PROVIDERS;
 
 import com.google.common.collect.Maps;
 import java.io.IOException;
@@ -100,7 +99,6 @@ public void startUp() throws Exception {
     conf.set(OSSProperties.GRAVITINO_OSS_ACCESS_KEY_SECRET, OSS_SECRET_KEY);
     conf.set(OSSProperties.GRAVITINO_OSS_ENDPOINT, OSS_ENDPOINT);
     conf.set("fs.oss.impl", "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem");
-    conf.set(FS_FILESYSTEM_PROVIDERS, "oss");
   }
 
   @AfterAll

clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/integration/test/GravitinoVirtualFileSystemS3IT.java

@@ -20,7 +20,6 @@
 package org.apache.gravitino.filesystem.hadoop.integration.test;
 
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS;
-import static org.apache.gravitino.filesystem.hadoop.GravitinoVirtualFileSystemConfiguration.FS_FILESYSTEM_PROVIDERS;
 
 import com.google.common.collect.Maps;
 import java.io.IOException;
@@ -157,7 +156,6 @@ public void startUp() throws Exception {
     conf.set(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, accessKey);
     conf.set(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, secretKey);
     conf.set(S3Properties.GRAVITINO_S3_ENDPOINT, s3Endpoint);
-    conf.set(FS_FILESYSTEM_PROVIDERS, "s3");
   }
 
   @AfterAll