[client] Allow FileSystem implementations expose customized options via 'client.fs.*' prefix (#938)

Author: luoyuxia

fluss-client/src/main/java/com/alibaba/fluss/client/FlussConnection.java

@@ -31,6 +31,7 @@
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.exception.FlussRuntimeException;
+import com.alibaba.fluss.fs.FileSystem;
 import com.alibaba.fluss.metadata.TableInfo;
 import com.alibaba.fluss.metadata.TablePath;
 import com.alibaba.fluss.metrics.registry.MetricRegistry;
@@ -40,9 +41,12 @@
 import com.alibaba.fluss.rpc.metrics.ClientMetricGroup;
 
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.List;
 
 import static com.alibaba.fluss.client.utils.MetadataUtils.getOneAvailableTabletServerNode;
+import static com.alibaba.fluss.config.FlussConfigUtils.CLIENT_PREFIX;
+import static com.alibaba.fluss.utils.PropertiesUtils.extractPrefix;
 
 /** A connection to Fluss cluster, and holds the client session resources. */
 public final class FlussConnection implements Connection {
@@ -63,6 +67,12 @@ public final class FlussConnection implements Connection {
 
     FlussConnection(Configuration conf, MetricRegistry metricRegistry) {
         this.conf = conf;
+        // init Filesystem with configuration from FlussConnection,
+        // only pass options with 'client.fs.' prefix
+        FileSystem.initialize(
+                Configuration.fromMap(
+                        extractPrefix(new HashMap<>(conf.toMap()), CLIENT_PREFIX + "fs.")),
+                null);
         // for client metrics.
         setupClientMetricsConfiguration();
         String clientId = conf.getString(ConfigOptions.CLIENT_ID);

fluss-client/src/test/java/com/alibaba/fluss/client/table/FlussTableITCase.java

@@ -31,6 +31,8 @@
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.config.MemorySize;
+import com.alibaba.fluss.fs.FsPath;
+import com.alibaba.fluss.fs.TestFileSystem;
 import com.alibaba.fluss.metadata.DataLakeFormat;
 import com.alibaba.fluss.metadata.KvFormat;
 import com.alibaba.fluss.metadata.LogFormat;
@@ -61,6 +63,7 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
@@ -1160,4 +1163,19 @@ void testMergeEngineWithVersion(boolean doProjection) throws Exception {
             }
         }
     }
+
+    @Test
+    void testFileSystemRecognizeConnectionConf() throws Exception {
+        Configuration config = new Configuration(clientConf);
+        config.setString("client.fs.test.key", "fs_test_value");
+        config.setString("client.test.key", "client_test_value");
+        try (Connection ignore = ConnectionFactory.createConnection(config)) {
+            FsPath fsPath = new FsPath("test:///f1");
+            TestFileSystem testFileSystem = (TestFileSystem) fsPath.getFileSystem();
+            Configuration filesystemConf = testFileSystem.getConfiguration();
+            assertThat(filesystemConf.toMap())
+                    .containsExactlyEntriesOf(
+                            Collections.singletonMap("client.fs.test.key", "fs_test_value"));
+        }
+    }
 }

fluss-common/src/test/java/com/alibaba/fluss/fs/TestFileSystem.java

@@ -44,6 +44,16 @@ public class TestFileSystem extends LocalFileSystem {
     private static final Map<FsPath, Integer> currentUnclosedOutputStream =
             MapUtils.newConcurrentHashMap();
 
+    private final Configuration configuration;
+
+    public TestFileSystem(Configuration configuration) {
+        this.configuration = configuration;
+    }
+
+    public Configuration getConfiguration() {
+        return configuration;
+    }
+
     public static int getNumtimeStreamOpened() {
         return streamOpenCounter.get();
     }
@@ -143,7 +153,7 @@ public String getScheme() {
 
         @Override
         public FileSystem create(URI fsUri, Configuration configuration) throws IOException {
-            return new TestFileSystem();
+            return new TestFileSystem(configuration);
         }
     }
 }

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/catalog/FlinkTableFactory.java

@@ -47,6 +47,7 @@
 
 import java.io.File;
 import java.time.ZoneId;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
@@ -82,11 +83,9 @@ public DynamicTableSource createDynamicTableSource(Context context) {
         FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
         final ReadableConfig tableOptions = helper.getOptions();
         Optional<DataLakeFormat> datalakeFormat = getDatalakeFormat(tableOptions);
-        if (datalakeFormat.isPresent()) {
-            helper.validateExcept("table.", "client.", datalakeFormat.get() + ".");
-        } else {
-            helper.validateExcept("table.", "client.");
-        }
+        List<String> prefixesToSkip = new ArrayList<>(Arrays.asList("table.", "client."));
+        datalakeFormat.ifPresent(dataLakeFormat -> prefixesToSkip.add(dataLakeFormat + "."));
+        helper.validateExcept(prefixesToSkip.toArray(new String[0]));
 
         boolean isStreamingMode =
                 context.getConfiguration().get(ExecutionOptions.RUNTIME_MODE)

fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java

@@ -145,7 +145,14 @@ void testPkTableReadOnlySnapshot() throws Exception {
         List<String> expectedRows = Arrays.asList("+I[1, v1]", "+I[2, v2]", "+I[3, v3]");
 
         assertResultsIgnoreOrder(
-                tEnv.executeSql("select * from read_snapshot_test").collect(), expectedRows, true);
+                tEnv.executeSql(
+                                // the options is just used to check option with prefix 'client.fs'
+                                // should
+                                // pass Flink validation
+                                "select * from read_snapshot_test /*+ OPTIONS('client.fs.oss.endpoint' = 'test') */")
+                        .collect(),
+                expectedRows,
+                true);
     }
 
     @Test