rename

Author: xx789633

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

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

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

@@ -30,6 +30,6 @@ public FlinkCatalog createCatalog(Context context) {
                 catalog.bootstrapServers,
                 catalog.classLoader,
                 catalog.securityConfigs,
-                catalog.lakeCatalogProperties);
+                catalog.lakeCatalogPropertiesSupplier);
     }
 }

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

@@ -43,7 +43,7 @@ static void beforeAll() {
                         catalog.bootstrapServers,
                         catalog.classLoader,
                         catalog.securityConfigs,
-                        catalog.lakeCatalogProperties);
+                        catalog.lakeCatalogPropertiesSupplier);
         catalog.open();
     }
 

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

@@ -116,7 +116,7 @@ public class FlinkCatalog extends AbstractCatalog {
     protected final String bootstrapServers;
     protected final Map<String, String> securityConfigs;
     protected final LakeFlinkCatalog lakeFlinkCatalog;
-    protected final Supplier<Map<String, String>> lakeCatalogProperties;
+    protected final Supplier<Map<String, String>> lakeCatalogPropertiesSupplier;
     protected Connection connection;
     protected Admin admin;
 
@@ -126,14 +126,14 @@ public FlinkCatalog(
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs,
-            Supplier<Map<String, String>> lakeCatalogProperties) {
+            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {
         this(
                 name,
                 defaultDatabase,
                 bootstrapServers,
                 classLoader,
                 securityConfigs,
-                lakeCatalogProperties,
+                lakeCatalogPropertiesSupplier,
                 new LakeFlinkCatalog(name, classLoader));
     }
 
@@ -144,15 +144,15 @@ public FlinkCatalog(
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs,
-            Supplier<Map<String, String>> lakeCatalogProperties,
+            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier,
             LakeFlinkCatalog lakeFlinkCatalog) {
         super(name, defaultDatabase);
         this.catalogName = name;
         this.defaultDatabase = defaultDatabase;
         this.bootstrapServers = bootstrapServers;
         this.classLoader = classLoader;
         this.securityConfigs = securityConfigs;
-        this.lakeCatalogProperties = lakeCatalogProperties;
+        this.lakeCatalogPropertiesSupplier = lakeCatalogPropertiesSupplier;
         this.lakeFlinkCatalog = lakeFlinkCatalog;
     }
 
@@ -323,7 +323,7 @@ public CatalogBaseTable getTable(ObjectPath objectPath)
                         objectPath.getDatabaseName(),
                         tableName,
                         tableInfo.getProperties(),
-                        lakeCatalogProperties);
+                        lakeCatalogPropertiesSupplier);
             } else {
                 tableInfo = admin.getTableInfo(tablePath).get();
             }
@@ -360,7 +360,7 @@ protected CatalogBaseTable getLakeTable(
             String databaseName,
             String tableName,
             Configuration properties,
-            Supplier<Map<String, String>> lakeCatalogProperties)
+            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier)
             throws TableNotExistException, CatalogException {
         String[] tableComponents = tableName.split("\\" + LAKE_TABLE_SPLITTER);
         if (tableComponents.length == 1) {
@@ -372,7 +372,7 @@ protected CatalogBaseTable getLakeTable(
             tableName = String.join("", tableComponents);
         }
         return lakeFlinkCatalog
-                .getLakeCatalog(properties, lakeCatalogProperties)
+                .getLakeCatalog(properties, lakeCatalogPropertiesSupplier)
                 .getTable(new ObjectPath(databaseName, tableName));
     }
 
@@ -787,7 +787,7 @@ public Map<String, String> getSecurityConfigs() {
     }
 
     @VisibleForTesting
-    public Supplier<Map<String, String>> getLakeCatalogProperties() {
-        return lakeCatalogProperties;
+    public Map<String, String> getLakeCatalogProperties() {
+        return lakeCatalogPropertiesSupplier.get();
     }
 }

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

@@ -52,7 +52,8 @@ public LakeFlinkCatalog(String catalogName, ClassLoader classLoader) {
     }
 
     public Catalog getLakeCatalog(
-            Configuration tableOptions, Supplier<Map<String, String>> lakeCatalogProperties) {
+            Configuration tableOptions,
+            Supplier<Map<String, String>> lakeCatalogPropertiesSupplier) {
         // TODO: Currently, a Fluss cluster only supports a single DataLake storage.
         // However, in the
         //  future, it may support multiple DataLakes. The following code assumes
@@ -74,7 +75,7 @@ public Catalog getLakeCatalog(
                     }
                     Map<String, String> catalogProperties =
                             PropertiesUtils.extractAndRemovePrefix(
-                                    lakeCatalogProperties.get(), lakeFormat + ".");
+                                    lakeCatalogPropertiesSupplier.get(), lakeFormat + ".");
 
                     catalogProperties.putAll(
                             DataLakeUtils.extractLakeCatalogProperties(tableOptions));

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().get()).isEqualTo(lakeCatalogMap);
+        assertThat(actualCatalog2.getLakeCatalogProperties()).isEqualTo(lakeCatalogMap);
     }
 
     @Test

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

@@ -829,7 +829,7 @@ void testCreateCatalogWithLakeProperties() {
         FlinkCatalog catalog =
                 (FlinkCatalog) tEnv.getCatalog("test_catalog_with_lake_properties").get();
 
-        assertOptionsEqual(catalog.getLakeCatalogProperties().get(), properties);
+        assertOptionsEqual(catalog.getLakeCatalogProperties(), properties);
     }
 
     /**

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

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