[#5991] feat(gcs): unify the GCS server acount path configuration for…

… fileset and GCSCredentialProvider (#5992)

### What changes were proposed in this pull request?

fileset use `gcs-service-account-file` while gcsTokenCredentialProvider
use `gcs-credential-file-path`, we'd better unify the name, use
`gcs-service-account-file` for GCS credential to unify them.

### Why are the changes needed?

Fix: #5991 

### Does this PR introduce _any_ user-facing change?
yes

### How was this patch tested?
existing tests

bundles/gcp/src/main/java/org/apache/gravitino/gcs/fs/GCSFileSystemProvider.java

@@ -35,7 +35,8 @@ public class GCSFileSystemProvider implements FileSystemProvider {
 
   @VisibleForTesting
   public static final Map<String, String> GRAVITINO_KEY_TO_GCS_HADOOP_KEY =
-      ImmutableMap.of(GCSProperties.GCS_SERVICE_ACCOUNT_JSON_PATH, GCS_SERVICE_ACCOUNT_JSON_FILE);
+      ImmutableMap.of(
+          GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, GCS_SERVICE_ACCOUNT_JSON_FILE);
 
   @Override
   public FileSystem getFileSystem(Path path, Map<String, String> config) throws IOException {

catalogs/catalog-common/src/main/java/org/apache/gravitino/storage/GCSProperties.java

@@ -22,7 +22,7 @@
 public class GCSProperties {
 
   // The path of service account JSON file of Google Cloud Storage.
-  public static final String GCS_SERVICE_ACCOUNT_JSON_PATH = "gcs-service-account-file";
+  public static final String GRAVITINO_GCS_SERVICE_ACCOUNT_FILE = "gcs-service-account-file";
 
   private GCSProperties() {}
 }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopGCSCatalogIT.java

@@ -19,7 +19,7 @@
 package org.apache.gravitino.catalog.hadoop.integration.test;
 
 import static org.apache.gravitino.catalog.hadoop.HadoopCatalogPropertiesMetadata.FILESYSTEM_PROVIDERS;
-import static org.apache.gravitino.storage.GCSProperties.GCS_SERVICE_ACCOUNT_JSON_PATH;
+import static org.apache.gravitino.storage.GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE;
 
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
@@ -99,7 +99,7 @@ protected String defaultBaseLocation() {
 
   protected void createCatalog() {
     Map<String, String> map = Maps.newHashMap();
-    map.put(GCS_SERVICE_ACCOUNT_JSON_PATH, SERVICE_ACCOUNT_FILE);
+    map.put(GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, SERVICE_ACCOUNT_FILE);
     map.put(FILESYSTEM_PROVIDERS, "gcs");
     metalake.createCatalog(catalogName, Catalog.Type.FILESET, provider, "comment", map);
 
@@ -117,7 +117,7 @@ public void testCreateSchemaAndFilesetWithSpecialLocation() {
     String ossLocation = String.format("gs://%s", BUCKET_NAME);
     Map<String, String> catalogProps = Maps.newHashMap();
     catalogProps.put("location", ossLocation);
-    catalogProps.put(GCS_SERVICE_ACCOUNT_JSON_PATH, SERVICE_ACCOUNT_FILE);
+    catalogProps.put(GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, SERVICE_ACCOUNT_FILE);
     catalogProps.put(FILESYSTEM_PROVIDERS, "gcs");
 
     Catalog localCatalog =

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

@@ -90,7 +90,7 @@ public void startUp() throws Exception {
     conf.set("fs.gravitino.client.metalake", metalakeName);
 
     // Pass this configuration to the real file system
-    conf.set(GCSProperties.GCS_SERVICE_ACCOUNT_JSON_PATH, SERVICE_ACCOUNT_FILE);
+    conf.set(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE, SERVICE_ACCOUNT_FILE);
   }
 
   @AfterAll

core/src/main/java/org/apache/gravitino/credential/config/GCSCredentialConfig.java

@@ -19,21 +19,18 @@
 
 package org.apache.gravitino.credential.config;
 
-import com.google.common.annotations.VisibleForTesting;
 import java.util.Map;
 import javax.annotation.Nullable;
 import org.apache.gravitino.Config;
 import org.apache.gravitino.config.ConfigBuilder;
 import org.apache.gravitino.config.ConfigConstants;
 import org.apache.gravitino.config.ConfigEntry;
+import org.apache.gravitino.storage.GCSProperties;
 
 public class GCSCredentialConfig extends Config {
 
-  @VisibleForTesting
-  public static final String GRAVITINO_GCS_CREDENTIAL_FILE_PATH = "gcs-credential-file-path";
-
   public static final ConfigEntry<String> GCS_CREDENTIAL_FILE_PATH =
-      new ConfigBuilder(GRAVITINO_GCS_CREDENTIAL_FILE_PATH)
+      new ConfigBuilder(GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE)
           .doc("The path of GCS credential file")
           .version(ConfigConstants.VERSION_0_7_0)
           .stringConf()

dev/docker/iceberg-rest-server/rewrite_config.py

@@ -24,7 +24,8 @@
   "GRAVITINO_WAREHOUSE" : "warehouse",
   "GRAVITINO_CREDENTIAL_PROVIDER_TYPE" : "credential-providers",
   "GRAVITINO_CREDENTIAL_PROVIDERS" : "credential-providers",
-  "GRAVITINO_GCS_CREDENTIAL_FILE_PATH" : "gcs-credential-file-path",
+  "GRAVITINO_GCS_CREDENTIAL_FILE_PATH" : "gcs-service-account-file",
+  "GRAVITINO_GCS_SERVICE_ACCOUNT_FILE" : "gcs-service-account-file",
   "GRAVITINO_S3_ACCESS_KEY" : "s3-access-key-id",
   "GRAVITINO_S3_SECRET_KEY" : "s3-secret-access-key",
   "GRAVITINO_S3_REGION" : "s3-region",

docs/how-to-use-gvfs.md

@@ -68,11 +68,11 @@ Apart from the above properties, to access fileset like S3, GCS, OSS and custom
 
 #### S3 fileset
 
-| Configuration item             | Description                                                                                                                                                                            | Default value | Required                 | Since version    |
-|--------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|--------------------------|------------------|
-| `s3-endpoint`                  | The endpoint of the AWS S3.                                                                                                                                                            | (none)        | Yes if it's a S3 fileset.| 0.7.0-incubating |
-| `s3-access-key-id`             | The access key of the AWS S3.                                                                                                                                                          | (none)        | Yes if it's a S3 fileset.| 0.7.0-incubating |
-| `s3-secret-access-key`         | The secret key of the AWS S3.                                                                                                                                                          | (none)        | Yes if it's a S3 fileset.| 0.7.0-incubating |
+| Configuration item     | Description                   | Default value | Required                  | Since version    |
+|------------------------|-------------------------------|---------------|---------------------------|------------------|
+| `s3-endpoint`          | The endpoint of the AWS S3.   | (none)        | Yes if it's a S3 fileset. | 0.7.0-incubating |
+| `s3-access-key-id`     | The access key of the AWS S3. | (none)        | Yes if it's a S3 fileset. | 0.7.0-incubating |
+| `s3-secret-access-key` | The secret key of the AWS S3. | (none)        | Yes if it's a S3 fileset. | 0.7.0-incubating |
 
 At the same time, you need to add the corresponding bundle jar
 1. [`gravitino-aws-bundle-${version}.jar`](https://repo1.maven.org/maven2/org/apache/gravitino/gravitino-aws-bundle/) in the classpath if no hadoop environment is available, or
@@ -81,9 +81,9 @@ At the same time, you need to add the corresponding bundle jar
 
 #### GCS fileset
 
-| Configuration item             | Description                                                                                                                                                                              | Default value | Required                  | Since version    |
-|--------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------|---------------------------|------------------|
-| `gcs-service-account-file`     | The path of GCS service account JSON file.                                                                                                                                               | (none)        | Yes if it's a GCS fileset.| 0.7.0-incubating |
+| Configuration item         | Description                                | Default value | Required                   | Since version    |
+|----------------------------|--------------------------------------------|---------------|----------------------------|------------------|
+| `gcs-service-account-file` | The path of GCS service account JSON file. | (none)        | Yes if it's a GCS fileset. | 0.7.0-incubating |
 
 In the meantime, you need to add the corresponding bundle jar
 1. [`gravitino-gcp-bundle-${version}.jar`](https://repo1.maven.org/maven2/org/apache/gravitino/gravitino-gcp-bundle/) in the classpath if no hadoop environment is available, or

docs/iceberg-rest-service.md

@@ -162,7 +162,8 @@ Supports using static GCS credential file or generating GCS token to access GCS
 | `gravitino.iceberg-rest.io-impl`                  | The io implementation for `FileIO` in Iceberg, use `org.apache.iceberg.gcp.gcs.GCSFileIO` for GCS. | (none)        | No       | 0.6.0-incubating |
 | `gravitino.iceberg-rest.credential-provider-type`  | Deprecated, please use `gravitino.iceberg-rest.credential-providers` instead.                      | (none)        | No                                                 | 0.7.0-incubating |
 | `gravitino.iceberg-rest.credential-providers`     | Supports `gcs-token`, generates a temporary token according to the query data path.                | (none)        | No       | 0.7.0-incubating |
-| `gravitino.iceberg-rest.gcs-credential-file-path` | The location of GCS credential file, only used when `credential-providers` is `gcs-token`.         | (none)        | No       | 0.7.0-incubating |
+| `gravitino.iceberg-rest.gcs-credential-file-path` | Deprecated, please use `gravitino.iceberg-rest.gcs-service-account-file` instead.                  | (none)        | No       | 0.7.0-incubating |
+| `gravitino.iceberg-rest.gcs-service-account-file` | The location of GCS credential file, only used when `credential-provider-type` is `gcs-token`.     | (none)        | No       | 0.8.0-incubating |
 
 For other Iceberg GCS properties not managed by Gravitino like `gcs.project-id`, you could config it directly by `gravitino.iceberg-rest.gcs.project-id`.
 
@@ -450,9 +451,8 @@ Gravitino Iceberg REST server in docker image could access local storage by defa
 | `GRAVITINO_IO_IMPL`                    | `gravitino.iceberg-rest.io-impl`                    | 0.7.0-incubating |
 | `GRAVITINO_URI`                        | `gravitino.iceberg-rest.uri`                        | 0.7.0-incubating |
 | `GRAVITINO_WAREHOUSE`                  | `gravitino.iceberg-rest.warehouse`                  | 0.7.0-incubating |
-| `GRAVITINO_CREDENTIAL_PROVIDER_TYPE`   | `gravitino.iceberg-rest.credential-providers`       | 0.8.0-incubating |
 | `GRAVITINO_CREDENTIAL_PROVIDERS`       | `gravitino.iceberg-rest.credential-providers`       | 0.8.0-incubating |
-| `GRAVITINO_GCS_CREDENTIAL_FILE_PATH`   | `gravitino.iceberg-rest.gcs-credential-file-path`   | 0.7.0-incubating |
+| `GRAVITINO_GCS_SERVICE_ACCOUNT_FILE`   | `gravitino.iceberg-rest.gcs-service-account-file`   | 0.8.0-incubating |
 | `GRAVITINO_S3_ACCESS_KEY`              | `gravitino.iceberg-rest.s3-access-key-id`           | 0.7.0-incubating |
 | `GRAVITINO_S3_SECRET_KEY`              | `gravitino.iceberg-rest.s3-secret-access-key`       | 0.7.0-incubating |
 | `GRAVITINO_S3_REGION`                  | `gravitino.iceberg-rest.s3-region`                  | 0.7.0-incubating |
@@ -465,6 +465,13 @@ Gravitino Iceberg REST server in docker image could access local storage by defa
 | `GRAVITINO_AZURE_CLIENT_ID`            | `gravitino.iceberg-rest.azure-client-id`            | 0.8.0-incubating |
 | `GRAVITINO_AZURE_CLIENT_SECRET`        | `gravitino.iceberg-rest.azure-client-secret`        | 0.8.0-incubating |
 
+The below environment is deprecated, please use the corresponding configuration items instead.
+
+| Deprecated Environment variables     | New environment variables            | Since version    | Deprecated version |
+|--------------------------------------|--------------------------------------|------------------|--------------------|
+| `GRAVITINO_CREDENTIAL_PROVIDER_TYPE` | `GRAVITINO_CREDENTIAL_PROVIDERS`     | 0.7.0-incubating | 0.8.0-incubating   |
+| `GRAVITINO_GCS_CREDENTIAL_FILE_PATH` | `GRAVITINO_GCS_SERVICE_ACCOUNT_FILE` | 0.7.0-incubating | 0.8.0-incubating   |
+
 Or build it manually to add custom configuration or logics:
 
 ```shell

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java

@@ -19,6 +19,8 @@
 
 package org.apache.gravitino.iceberg.service;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import java.util.Collections;
 import java.util.HashMap;
@@ -34,6 +36,7 @@
 import org.apache.gravitino.credential.PathBasedCredentialContext;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.common.ops.IcebergCatalogWrapper;
+import org.apache.gravitino.storage.GCSProperties;
 import org.apache.gravitino.utils.MapUtils;
 import org.apache.gravitino.utils.PrincipalUtils;
 import org.apache.iceberg.TableMetadata;
@@ -58,14 +61,23 @@ public class CatalogWrapperForREST extends IcebergCatalogWrapper {
           IcebergConstants.ICEBERG_S3_ENDPOINT,
           IcebergConstants.ICEBERG_OSS_ENDPOINT);
 
+  @SuppressWarnings("deprecation")
+  private static Map<String, String> deprecatedProperties =
+      ImmutableMap.of(
+          CredentialConstants.CREDENTIAL_PROVIDER_TYPE,
+          CredentialConstants.CREDENTIAL_PROVIDERS,
+          "gcs-credential-file-path",
+          GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE);
+
   public CatalogWrapperForREST(String catalogName, IcebergConfig config) {
     super(config);
     this.catalogConfigToClients =
         MapUtils.getFilteredMap(
             config.getIcebergCatalogProperties(),
             key -> catalogPropertiesToClientKeys.contains(key));
     // To be compatible with old properties
-    Map<String, String> catalogProperties = checkForCompatibility(config.getAllConfig());
+    Map<String, String> catalogProperties =
+        checkForCompatibility(config.getAllConfig(), deprecatedProperties);
     this.catalogCredentialManager = new CatalogCredentialManager(catalogName, catalogProperties);
   }
 
@@ -131,27 +143,30 @@ private LoadTableResponse injectCredentialConfig(
         .build();
   }
 
-  @SuppressWarnings("deprecation")
-  private Map<String, String> checkForCompatibility(Map<String, String> properties) {
-    HashMap<String, String> normalizedProperties = new HashMap<>(properties);
-    String credentialProviderType = properties.get(CredentialConstants.CREDENTIAL_PROVIDER_TYPE);
-    String credentialProviders = properties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
-    if (StringUtils.isNotBlank(credentialProviders)
-        && StringUtils.isNotBlank(credentialProviderType)) {
+  @VisibleForTesting
+  static Map<String, String> checkForCompatibility(
+      Map<String, String> properties, Map<String, String> deprecatedProperties) {
+    Map<String, String> newProperties = new HashMap<>(properties);
+    deprecatedProperties.forEach(
+        (deprecatedProperty, newProperty) -> {
+          replaceDeprecatedProperties(newProperties, deprecatedProperty, newProperty);
+        });
+    return newProperties;
+  }
+
+  private static void replaceDeprecatedProperties(
+      Map<String, String> properties, String deprecatedProperty, String newProperty) {
+    String deprecatedValue = properties.get(deprecatedProperty);
+    String newValue = properties.get(newProperty);
+    if (StringUtils.isNotBlank(deprecatedValue) && StringUtils.isNotBlank(newValue)) {
       throw new IllegalArgumentException(
-          String.format(
-              "Should not set both %s and %s",
-              CredentialConstants.CREDENTIAL_PROVIDER_TYPE,
-              CredentialConstants.CREDENTIAL_PROVIDERS));
+          String.format("Should not set both %s and %s", deprecatedProperty, newProperty));
     }
 
-    if (StringUtils.isNotBlank(credentialProviderType)) {
-      LOG.warn(
-          "%s is deprecated, please use %s instead.",
-          CredentialConstants.CREDENTIAL_PROVIDER_TYPE, CredentialConstants.CREDENTIAL_PROVIDERS);
-      normalizedProperties.put(CredentialConstants.CREDENTIAL_PROVIDERS, credentialProviderType);
+    if (StringUtils.isNotBlank(deprecatedValue)) {
+      LOG.warn("%s is deprecated, please use %s instead.", deprecatedProperty, newProperty);
+      properties.remove(deprecatedProperty);
+      properties.put(newProperty, deprecatedValue);
     }
-
-    return normalizedProperties;
   }
 }

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/integration/test/IcebergRESTGCSIT.java

@@ -25,11 +25,11 @@
 import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
 import org.apache.gravitino.credential.CredentialConstants;
 import org.apache.gravitino.credential.GCSTokenCredential;
-import org.apache.gravitino.credential.config.GCSCredentialConfig;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.integration.test.util.BaseIT;
 import org.apache.gravitino.integration.test.util.DownloaderUtils;
 import org.apache.gravitino.integration.test.util.ITUtils;
+import org.apache.gravitino.storage.GCSProperties;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
 
 // You should export GRAVITINO_GCS_BUCKET and GOOGLE_APPLICATION_CREDENTIALS to run the test
@@ -76,8 +76,7 @@ private Map<String, String> getGCSConfig() {
         IcebergConfig.ICEBERG_CONFIG_PREFIX + CredentialConstants.CREDENTIAL_PROVIDERS,
         GCSTokenCredential.GCS_TOKEN_CREDENTIAL_TYPE);
     configMap.put(
-        IcebergConfig.ICEBERG_CONFIG_PREFIX
-            + GCSCredentialConfig.GRAVITINO_GCS_CREDENTIAL_FILE_PATH,
+        IcebergConfig.ICEBERG_CONFIG_PREFIX + GCSProperties.GRAVITINO_GCS_SERVICE_ACCOUNT_FILE,
         gcsCredentialPath);
     configMap.put(
         IcebergConfig.ICEBERG_CONFIG_PREFIX + IcebergConstants.IO_IMPL,

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/TestCatalogWrapperForREST.java

@@ -0,0 +1,49 @@
+/*
+ *  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.iceberg.service;
+
+import java.util.Map;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
+
+public class TestCatalogWrapperForREST {
+
+  @Test
+  void testCheckPropertiesForCompatibility() {
+    ImmutableMap<String, String> deprecatedMap = ImmutableMap.of("deprecated", "new");
+    ImmutableMap<String, String> propertiesWithDeprecatedKey = ImmutableMap.of("deprecated", "v");
+    Map<String, String> newProperties =
+        CatalogWrapperForREST.checkForCompatibility(propertiesWithDeprecatedKey, deprecatedMap);
+    Assertions.assertEquals(newProperties, ImmutableMap.of("new", "v"));
+
+    ImmutableMap<String, String> propertiesWithoutDeprecatedKey = ImmutableMap.of("k", "v");
+    newProperties =
+        CatalogWrapperForREST.checkForCompatibility(propertiesWithoutDeprecatedKey, deprecatedMap);
+    Assertions.assertEquals(newProperties, ImmutableMap.of("k", "v"));
+
+    ImmutableMap<String, String> propertiesWithBothKey =
+        ImmutableMap.of("deprecated", "v", "new", "v");
+
+    Assertions.assertThrowsExactly(
+        IllegalArgumentException.class,
+        () -> CatalogWrapperForREST.checkForCompatibility(propertiesWithBothKey, deprecatedMap));
+  }
+}