rebase

Author: xx789633

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

@@ -61,13 +61,15 @@ public Flink21Catalog(
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogProperties,
             LakeFlinkCatalog lakeFlinkCatalog) {
         super(
                 name,
                 defaultDatabase,
                 bootstrapServers,
                 classLoader,
                 securityConfigs,
+                lakeCatalogProperties,
                 lakeFlinkCatalog);
     }
 

fluss-flink/fluss-flink-2.1/src/test/java/org/apache/fluss/flink/catalog/FlinkCatalog21Test.java

@@ -43,6 +43,7 @@ protected FlinkCatalog initCatalog(
                 bootstrapServers,
                 Thread.currentThread().getContextClassLoader(),
                 Collections.emptyMap(),
+                Collections::emptyMap,
                 lakeFlinkCatalog);
     }
 

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

@@ -125,13 +125,15 @@ public FlinkCatalog(
             String defaultDatabase,
             String bootstrapServers,
             ClassLoader classLoader,
-            Map<String, String> securityConfigs) {
+            Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogProperties) {
         this(
                 name,
                 defaultDatabase,
                 bootstrapServers,
                 classLoader,
                 securityConfigs,
+                lakeCatalogProperties,
                 new LakeFlinkCatalog(name, classLoader));
     }
 
@@ -142,13 +144,15 @@ public FlinkCatalog(
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs,
+            Supplier<Map<String, String>> lakeCatalogProperties,
             LakeFlinkCatalog lakeFlinkCatalog) {
         super(name, defaultDatabase);
         this.catalogName = name;
         this.defaultDatabase = defaultDatabase;
         this.bootstrapServers = bootstrapServers;
         this.classLoader = classLoader;
         this.securityConfigs = securityConfigs;
+        this.lakeCatalogProperties = lakeCatalogProperties;
         this.lakeFlinkCatalog = lakeFlinkCatalog;
     }
 

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

@@ -72,6 +72,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 import static org.apache.fluss.config.ConfigOptions.BOOTSTRAP_SERVERS;
 import static org.apache.fluss.config.ConfigOptions.TABLE_DATALAKE_ENABLED;
@@ -172,6 +173,7 @@ protected FlinkCatalog initCatalog(
                 bootstrapServers,
                 Thread.currentThread().getContextClassLoader(),
                 Collections.emptyMap(),
+                Collections::emptyMap,
                 lakeFlinkCatalog);
     }
 
@@ -1009,7 +1011,8 @@ public MockLakeFlinkCatalog(String catalogName, ClassLoader classLoader) {
         }
 
         @Override
-        public Catalog getLakeCatalog(Configuration tableOptions) {
+        public Catalog getLakeCatalog(
+                Configuration tableOptions, Supplier<Map<String, String>> lakeCatalogProperties) {
             return catalog;
         }