[lake] Remove sensetive lake catalog properties in getTable (apache#1860)

Author: xx789633

fluss-client/src/test/java/org/apache/fluss/client/admin/ClientToServerITCaseBase.java

@@ -117,6 +117,9 @@ private static Configuration initConfig() {
         // set default datalake format for the cluster and enable datalake tables
         conf.set(ConfigOptions.DATALAKE_FORMAT, DataLakeFormat.PAIMON);
 
+        conf.setString("datalake.paimon.jdbc.user", "admin");
+        conf.setString("datalake.paimon.jdbc.password", "pass");
+
         conf.set(ConfigOptions.CLIENT_WRITER_BUFFER_MEMORY_SIZE, MemorySize.parse("1mb"));
         conf.set(ConfigOptions.CLIENT_WRITER_BATCH_SIZE, MemorySize.parse("1kb"));
         conf.set(ConfigOptions.MAX_PARTITION_NUM, 10);

fluss-client/src/test/java/org/apache/fluss/client/admin/FlussAdminITCase.java

@@ -208,6 +208,23 @@ void testGetTableInfoAndSchema() throws Exception {
         // assert created time
         assertThat(tableInfo.getCreatedTime())
                 .isBetween(timestampBeforeCreate, timestampAfterCreate);
+
+        // test sensitive lake catalog properties have been removed
+        tablePath = TablePath.of("test_db", "lake_table");
+        admin.createTable(
+                        tablePath,
+                        DEFAULT_TABLE_DESCRIPTOR.withProperties(
+                                new HashMap<String, String>() {
+                                    {
+                                        put("table.datalake.enabled", "true");
+                                    }
+                                }),
+                        false)
+                .get();
+        Map<String, String> properties =
+                admin.getTableInfo(tablePath).get().getProperties().toMap();
+        assertThat(properties.containsKey("table.datalake.paimon.jdbc.user")).isTrue();
+        assertThat(properties.containsKey("table.datalake.paimon.jdbc.password")).isFalse();
     }
 
     @Test

fluss-server/src/main/java/org/apache/fluss/server/coordinator/MetadataManager.java

@@ -62,6 +62,8 @@
 
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -81,6 +83,14 @@ public class MetadataManager {
     private final int maxBucketNum;
     private final LakeCatalogDynamicLoader lakeCatalogDynamicLoader;
 
+    public static final Set<String> SENSITIVE_TABLE_OPTIOINS = new HashSet<>();
+
+    static {
+        SENSITIVE_TABLE_OPTIOINS.add("password");
+        SENSITIVE_TABLE_OPTIOINS.add("secret");
+        SENSITIVE_TABLE_OPTIOINS.add("key");
+    }
+
     /**
      * Creates a new metadata manager.
      *
@@ -507,6 +517,20 @@ private boolean isDataLakeEnabled(Map<String, String> properties) {
         return Boolean.parseBoolean(dataLakeEnabledValue);
     }
 
+    public void removeSensitiveTableOptions(Map<String, String> tableLakeOptions) {
+        if (tableLakeOptions == null || tableLakeOptions.isEmpty()) {
+            return;
+        }
+
+        Iterator<Map.Entry<String, String>> iterator = tableLakeOptions.entrySet().iterator();
+        while (iterator.hasNext()) {
+            String key = iterator.next().getKey().toLowerCase();
+            if (SENSITIVE_TABLE_OPTIOINS.stream().anyMatch(key::contains)) {
+                iterator.remove();
+            }
+        }
+    }
+
     public TableInfo getTable(TablePath tablePath) throws TableNotExistException {
         Optional<TableRegistration> optionalTable;
         try {
@@ -520,10 +544,10 @@ public TableInfo getTable(TablePath tablePath) throws TableNotExistException {
         }
         TableRegistration tableReg = optionalTable.get();
         SchemaInfo schemaInfo = getLatestSchema(tablePath);
-        return tableReg.toTableInfo(
-                tablePath,
-                schemaInfo,
-                lakeCatalogDynamicLoader.getLakeCatalogContainer().getDefaultTableLakeOptions());
+        Map<String, String> tableLakeOptions =
+                lakeCatalogDynamicLoader.getLakeCatalogContainer().getDefaultTableLakeOptions();
+        removeSensitiveTableOptions(tableLakeOptions);
+        return tableReg.toTableInfo(tablePath, schemaInfo, tableLakeOptions);
     }
 
     public Map<TablePath, TableInfo> getTables(Collection<TablePath> tablePaths)