[gs] Map gs.* config entries to fs.gs.* Hadoop config keys

paimon-filesystems/paimon-gs-impl/src/main/java/org/apache/paimon/gs/GSFileIO.java

@@ -18,10 +18,10 @@
 
 package org.apache.paimon.gs;
 
+import org.apache.paimon.annotation.VisibleForTesting;
 import org.apache.paimon.catalog.CatalogContext;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.options.Options;
-import org.apache.paimon.utils.SensitiveConfigUtils;
 
 import com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem;
 import org.apache.hadoop.conf.Configuration;
@@ -46,6 +46,8 @@ public class GSFileIO extends HadoopCompliantFileIO {
 
     private static final String[] CONFIG_PREFIXES = {"gs.", "fs.gs."};
 
+    private static final String HADOOP_CONFIG_PREFIX = "fs.gs.";
+
     /**
      * Cache GSFileSystem, at present, there is no good mechanism to ensure that the file system
      * will be shut down, so here the fs cache is used to avoid resource leakage.
@@ -61,20 +63,25 @@ public boolean isObjectStore() {
 
     @Override
     public void configure(CatalogContext context) {
-        hadoopOptions = new Options();
-        // read all configuration with prefix 'CONFIG_PREFIXES'
+        this.hadoopOptions = loadHadoopConfigFromContext(context);
+    }
+
+    // add additional config entries from the IO config to the Hadoop config
+    @VisibleForTesting
+    Options loadHadoopConfigFromContext(CatalogContext context) {
+        Options hadoopConfig = new Options();
         for (String key : context.options().keySet()) {
             for (String prefix : CONFIG_PREFIXES) {
                 if (key.startsWith(prefix)) {
+                    String newKey = HADOOP_CONFIG_PREFIX + key.substring(prefix.length());
                     String value = context.options().get(key);
-                    hadoopOptions.set(key, value);
-                    LOG.warn(
-                            "Adding config entry for {} as {} to Hadoop config",
-                            key,
-                            SensitiveConfigUtils.redactValue(key, hadoopOptions.get(key)));
+                    hadoopConfig.set(newKey, value);
+
+                    LOG.debug("Adding config entry for {} as {} to Hadoop config", key, newKey);
                 }
             }
         }
+        return hadoopConfig;
     }
 
     @Override

paimon-filesystems/paimon-gs-impl/src/test/java/org/apache/paimon/gs/GSFileIOTest.java

@@ -0,0 +1,66 @@
+/*
+ * 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.paimon.gs;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.options.Options;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link GSFileIO}. */
+public class GSFileIOTest {
+
+    @Test
+    public void testGsPrefixedKeyIsMappedToHadoopKey() {
+        Options options = new Options();
+        options.set("gs.auth.type", "SERVICE_ACCOUNT_JSON_KEYFILE");
+
+        Options hadoopConfig = loadHadoopConfig(options);
+
+        assertThat(hadoopConfig.get("fs.gs.auth.type")).isEqualTo("SERVICE_ACCOUNT_JSON_KEYFILE");
+        assertThat(hadoopConfig.containsKey("gs.auth.type")).isFalse();
+    }
+
+    @Test
+    public void testHadoopPrefixedKeyIsKept() {
+        Options options = new Options();
+        options.set("fs.gs.project.id", "my-project");
+
+        Options hadoopConfig = loadHadoopConfig(options);
+
+        assertThat(hadoopConfig.get("fs.gs.project.id")).isEqualTo("my-project");
+    }
+
+    @Test
+    public void testUnrelatedKeysAreIgnored() {
+        Options options = new Options();
+        options.set("warehouse", "gs://bucket/path");
+        options.set("s3.endpoint", "http://localhost:9000");
+
+        Options hadoopConfig = loadHadoopConfig(options);
+
+        assertThat(hadoopConfig.toMap()).isEmpty();
+    }
+
+    private static Options loadHadoopConfig(Options options) {
+        return new GSFileIO().loadHadoopConfigFromContext(CatalogContext.create(options));
+    }
+}