fix supplier

Author: xx789633

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

@@ -116,6 +116,7 @@ public class FlinkCatalog extends AbstractCatalog {
     protected final String bootstrapServers;
     protected final Map<String, String> securityConfigs;
     protected final LakeFlinkCatalog lakeFlinkCatalog;
+    protected volatile Map<String, String> lakeCatalogProperties;
     protected final Supplier<Map<String, String>> lakeCatalogPropertiesSupplier;
     protected Connection connection;
     protected Admin admin;
@@ -323,7 +324,7 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
                         objectPath.getDatabaseName(),
                         tableName,
                         tableInfo.getProperties(),
-                        lakeCatalogPropertiesSupplier);
+                        getLakeCatalogProperties());
             } else {
                 tableInfo = admin.getTableInfo(tablePath).get();
             }
@@ -360,7 +361,7 @@ protected CatalogBaseTable getLakeTable(
             String databaseName,
             String tableName,
             Configuration properties,
-            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier)
+            Map<String, String> lakeCatalogProperties)
             throws TableNotExistException, CatalogException {
         String[] tableComponents = tableName.split("\\" + LAKE_TABLE_SPLITTER);
         if (tableComponents.length == 1) {
@@ -372,7 +373,7 @@ protected CatalogBaseTable getLakeTable(
             tableName = String.join("", tableComponents);
         }
         return lakeFlinkCatalog
-                .getLakeCatalog(properties, lakeCatalogPropertiesSupplier)
+                .getLakeCatalog(properties, lakeCatalogProperties)
                 .getTable(new ObjectPath(databaseName, tableName));
     }
 
@@ -788,6 +789,13 @@ public Map<String, String> getSecurityConfigs() {
 
     @VisibleForTesting
     public Map<String, String> getLakeCatalogProperties() {
-        return lakeCatalogPropertiesSupplier.get();
+        if (lakeCatalogProperties == null) {
+            synchronized (this) {
+                if (lakeCatalogProperties == null) {
+                    lakeCatalogProperties = lakeCatalogPropertiesSupplier.get();
+                }
+            }
+        }
+        return lakeCatalogProperties;
     }
 }

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

@@ -31,7 +31,6 @@
 
 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;
@@ -52,8 +51,7 @@ public LakeFlinkCatalog(String catalogName, ClassLoader classLoader) {
     }
 
     public Catalog getLakeCatalog(
-            Configuration tableOptions,
-            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {
+            Configuration tableOptions, 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
@@ -75,7 +73,7 @@ public Catalog getLakeCatalog(
                     }
                     Map<String, String> catalogProperties =
                             PropertiesUtils.extractAndRemovePrefix(
-                                    lakeCatalogPropertiesSupplier.get(), lakeFormat + ".");
+                                    lakeCatalogProperties, 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/FlinkCatalogTest.java

@@ -72,7 +72,6 @@
 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;
@@ -1012,8 +1011,7 @@ public MockLakeFlinkCatalog(String catalogName, ClassLoader classLoader) {
 
         @Override
         public Catalog getLakeCatalog(
-                Configuration tableOptions,
-                Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {
+                Configuration tableOptions, Map<String, String> lakeCatalogProperties) {
             return catalog;
         }