supplier

Author: xx789633

fluss-flink/fluss-flink-2.1/src/main/java/org/apache/fluss/flink/catalog/Flink21Catalog.java

@@ -31,6 +31,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 /** A {@link FlinkCatalog} used for Flink 2.1. */
 public class Flink21Catalog extends FlinkCatalog {
@@ -41,7 +42,7 @@ public Flink21Catalog(
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs,
-            Map<String, String> lakeCatalogProperties) {
+            Supplier<Map<String, String>> lakeCatalogProperties) {
         super(
                 name,
                 defaultDatabase,

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalog.java

@@ -80,6 +80,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import static org.apache.flink.util.Preconditions.checkArgument;
@@ -115,7 +116,7 @@ public class FlinkCatalog extends AbstractCatalog {
     protected final String bootstrapServers;
     protected final Map<String, String> securityConfigs;
     protected final LakeFlinkCatalog lakeFlinkCatalog;
-    protected final Map<String, String> lakeCatalogProperties;
+    protected final Supplier<Map<String, String>> lakeCatalogProperties;
     protected Connection connection;
     protected Admin admin;
 
@@ -125,7 +126,7 @@ public FlinkCatalog(
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs,
-            Map<String, String> lakeCatalogProperties) {
+            Supplier<Map<String, String>> lakeCatalogProperties) {
         super(name, defaultDatabase);
         this.catalogName = name;
         this.defaultDatabase = defaultDatabase;
@@ -302,7 +303,7 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
                         objectPath.getDatabaseName(),
                         tableName,
                         tableInfo.getProperties(),
-                        lakeCatalogProperties);
+                        lakeCatalogProperties.get());
             } else {
                 tableInfo = admin.getTableInfo(tablePath).get();
             }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/catalog/FlinkCatalogFactory.java

@@ -71,17 +71,18 @@ public FlinkCatalog createCatalog(Context context) {
         Map<String, String> options = context.getOptions();
         Map<String, String> securityConfigs = extractPrefix(options, CLIENT_SECURITY_PREFIX);
 
-        Map<String, String> lakeCatalogProperties = new HashMap<>();
-        for (DataLakeFormat lakeFormat : DataLakeFormat.values()) {
-            lakeCatalogProperties.putAll(extractPrefix(options, lakeFormat.toString()));
-        }
-
         return new FlinkCatalog(
                 context.getName(),
                 helper.getOptions().get(FlinkCatalogOptions.DEFAULT_DATABASE),
                 helper.getOptions().get(FlinkConnectorOptions.BOOTSTRAP_SERVERS),
                 context.getClassLoader(),
                 securityConfigs,
-                lakeCatalogProperties);
+                () -> {
+                    Map<String, String> lakeCatalogProperties = new HashMap<>();
+                    for (DataLakeFormat lakeFormat : DataLakeFormat.values()) {
+                        lakeCatalogProperties.putAll(extractPrefix(options, lakeFormat.toString()));
+                    }
+                    return lakeCatalogProperties;
+                });
     }
 }

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogFactoryTest.java

@@ -65,7 +65,7 @@ public void testCreateCatalog() {
                         BOOTSTRAP_SERVERS_NAME,
                         Thread.currentThread().getContextClassLoader(),
                         Collections.emptyMap(),
-                        Collections.emptyMap());
+                        Collections::emptyMap);
 
         checkEquals(flinkCatalog, actualCatalog);
 

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogITCase.java

@@ -116,7 +116,7 @@ static void beforeAll() {
                         bootstrapServers,
                         Thread.currentThread().getContextClassLoader(),
                         Collections.emptyMap(),
-                        Collections.emptyMap());
+                        Collections::emptyMap);
         catalog.open();
     }
 
@@ -771,7 +771,7 @@ void testAuthentication() throws Exception {
                             bootstrapServers,
                             Thread.currentThread().getContextClassLoader(),
                             Collections.emptyMap(),
-                            Collections.emptyMap());
+                            Collections::emptyMap);
             Catalog finalAuthenticateCatalog = authenticateCatalog;
             assertThatThrownBy(finalAuthenticateCatalog::open)
                     .cause()
@@ -790,7 +790,7 @@ void testAuthentication() throws Exception {
                             bootstrapServers,
                             Thread.currentThread().getContextClassLoader(),
                             clientConfig,
-                            Collections.emptyMap());
+                            Collections::emptyMap);
             authenticateCatalog.open();
             assertThat(authenticateCatalog.listDatabases())
                     .containsExactlyInAnyOrderElementsOf(Collections.singletonList(DEFAULT_DB));

fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalogTest.java

@@ -169,7 +169,7 @@ static void beforeAll() {
                         String.join(",", flussConf.get(BOOTSTRAP_SERVERS)),
                         Thread.currentThread().getContextClassLoader(),
                         Collections.emptyMap(),
-                        Collections.emptyMap());
+                        Collections::emptyMap);
         catalog.open();
     }
 
@@ -625,7 +625,7 @@ void testDatabase() throws Exception {
                                                 flussConf.get(ConfigOptions.BOOTSTRAP_SERVERS)),
                                         Thread.currentThread().getContextClassLoader(),
                                         Collections.emptyMap(),
-                                        Collections.emptyMap()))
+                                        Collections::emptyMap))
                 .hasMessageContaining("defaultDatabase cannot be null or empty");
     }
 
@@ -814,7 +814,7 @@ void testConnectionFailureHandling() {
                         "invalid-bootstrap-server:9092",
                         Thread.currentThread().getContextClassLoader(),
                         Collections.emptyMap(),
-                        Collections.emptyMap());
+                        Collections::emptyMap);
 
         // Test open() throws proper exception
         assertThatThrownBy(() -> badCatalog.open())
@@ -945,7 +945,7 @@ void testSecurityConfigsIntegration() throws Exception {
                         String.join(",", flussConf.get(BOOTSTRAP_SERVERS)),
                         Thread.currentThread().getContextClassLoader(),
                         securityConfigs,
-                        Collections.emptyMap());
+                        Collections::emptyMap);
         securedCatalog.open();
 
         try {

fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/flink/FlinkCatalogLakeTest.java

@@ -103,7 +103,7 @@ public void buildCatalog() {
                         bootstrapServers,
                         Thread.currentThread().getContextClassLoader(),
                         Collections.emptyMap(),
-                        Collections.emptyMap());
+                        Collections::emptyMap);
         catalog.open();
     }
 }