fix

Author: xx789633

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

@@ -303,7 +303,7 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
                         objectPath.getDatabaseName(),
                         tableName,
                         tableInfo.getProperties(),
-                        lakeCatalogProperties.get());
+                        lakeCatalogProperties);
             } else {
                 tableInfo = admin.getTableInfo(tablePath).get();
             }
@@ -340,7 +340,7 @@ protected CatalogBaseTable getLakeTable(
             String databaseName,
             String tableName,
             Configuration properties,
-            Map<String, String> lakeCatalogProperties)
+            Supplier<Map<String, String>> lakeCatalogProperties)
             throws TableNotExistException, CatalogException {
         String[] tableComponents = tableName.split("\\" + LAKE_TABLE_SPLITTER);
         if (tableComponents.length == 1) {
@@ -767,7 +767,7 @@ public Map<String, String> getSecurityConfigs() {
     }
 
     @VisibleForTesting
-    public Map<String, String> getLakeCatalogProperties() {
+    public Supplier<Map<String, String>> getLakeCatalogProperties() {
         return lakeCatalogProperties;
     }
 }

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeFlinkCatalog.java

@@ -31,6 +31,7 @@
 
 import java.lang.reflect.Method;
 import java.util.Map;
+import java.util.function.Supplier;
 
 import static org.apache.fluss.metadata.DataLakeFormat.ICEBERG;
 import static org.apache.fluss.metadata.DataLakeFormat.PAIMON;
@@ -51,7 +52,7 @@ public LakeFlinkCatalog(String catalogName, ClassLoader classLoader) {
     }
 
     public Catalog getLakeCatalog(
-            Configuration tableOptions, Map<String, String> lakeCatalogProperties) {
+            Configuration tableOptions, Supplier<Map<String, String>> lakeCatalogProperties) {
         // TODO: Currently, a Fluss cluster only supports a single DataLake storage.
         // However, in the
         //  future, it may support multiple DataLakes. The following code assumes
@@ -73,7 +74,7 @@ public Catalog getLakeCatalog(
                     }
                     Map<String, String> catalogProperties =
                             PropertiesUtils.extractAndRemovePrefix(
-                                    lakeCatalogProperties, lakeFormat + ".");
+                                    lakeCatalogProperties.get(), lakeFormat + ".");
 
                     catalogProperties.putAll(
                             DataLakeUtils.extractLakeCatalogProperties(tableOptions));

fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/lake/LakeTableFactory.java

@@ -85,7 +85,7 @@ private DynamicTableSourceFactory getIcebergFactory() {
                     lakeFlinkCatalog.getLakeCatalog(
                             // we can pass empty configuration to get catalog
                             // since the catalog should already be initialized
-                            new Configuration(), Collections.emptyMap());
+                            new Configuration(), Collections::emptyMap);
 
             // Create FlinkDynamicTableFactory with the catalog
             Class<?> icebergFactoryClass =

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

@@ -91,7 +91,7 @@ public void testCreateCatalog() {
                                 Thread.currentThread().getContextClassLoader());
 
         assertThat(actualCatalog2.getSecurityConfigs()).isEqualTo(securityMap);
-        assertThat(actualCatalog2.getLakeCatalogProperties()).isEqualTo(lakeCatalogMap);
+        assertThat(actualCatalog2.getLakeCatalogProperties().get()).isEqualTo(lakeCatalogMap);
     }
 
     @Test