[core] Introduce PrivilegeManagerLoader to avoid hardcode FileBasedPrivilegeManager

paimon-core/src/main/java/org/apache/paimon/privilege/FileBasedPrivilegeManagerLoader.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.privilege;
+
+import org.apache.paimon.fs.FileIO;
+
+/** Loader for creating a {@link FileBasedPrivilegeManager}. */
+public class FileBasedPrivilegeManagerLoader implements PrivilegeManagerLoader {
+
+    private static final long serialVersionUID = 1L;
+
+    private final String warehouse;
+    private final FileIO fileIO;
+    private final String user;
+    private final String password;
+
+    public FileBasedPrivilegeManagerLoader(
+            String warehouse, FileIO fileIO, String user, String password) {
+        this.warehouse = warehouse;
+        this.fileIO = fileIO;
+        this.user = user;
+        this.password = password;
+    }
+
+    @Override
+    public FileBasedPrivilegeManager load() {
+        return new FileBasedPrivilegeManager(warehouse, fileIO, user, password);
+    }
+}

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegeManagerLoader.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.privilege;
+
+import java.io.Serializable;
+
+/** Loader for creating a {@link PrivilegeManager}. */
+@FunctionalInterface
+public interface PrivilegeManagerLoader extends Serializable {
+
+    PrivilegeManager load();
+}

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedCatalog.java

@@ -23,7 +23,6 @@
 import org.apache.paimon.catalog.DelegateCatalog;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.catalog.PropertyChange;
-import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.options.ConfigOption;
 import org.apache.paimon.options.ConfigOptions;
 import org.apache.paimon.options.Options;
@@ -47,37 +46,27 @@ public class PrivilegedCatalog extends DelegateCatalog {
                     .stringType()
                     .defaultValue(PrivilegeManager.PASSWORD_ANONYMOUS);
 
-    private final String warehouse;
-    private final FileIO fileIO;
-    private final String user;
-    private final String password;
-
     private final PrivilegeManager privilegeManager;
+    private final PrivilegeManagerLoader privilegeManagerLoader;
 
-    public PrivilegedCatalog(
-            Catalog wrapped, String warehouse, FileIO fileIO, String user, String password) {
+    public PrivilegedCatalog(Catalog wrapped, PrivilegeManagerLoader privilegeManagerLoader) {
         super(wrapped);
-        this.warehouse = warehouse;
-        this.fileIO = fileIO;
-        this.user = user;
-        this.password = password;
-        this.privilegeManager = new FileBasedPrivilegeManager(warehouse, fileIO, user, password);
+        this.privilegeManager = privilegeManagerLoader.load();
+        this.privilegeManagerLoader = privilegeManagerLoader;
     }
 
     public static Catalog tryToCreate(Catalog catalog, Options options) {
-        if (new FileBasedPrivilegeManager(
+        FileBasedPrivilegeManagerLoader fileBasedPrivilegeManagerLoader =
+                new FileBasedPrivilegeManagerLoader(
                         catalog.warehouse(),
                         catalog.fileIO(),
                         options.get(PrivilegedCatalog.USER),
-                        options.get(PrivilegedCatalog.PASSWORD))
-                .privilegeEnabled()) {
-            catalog =
-                    new PrivilegedCatalog(
-                            catalog,
-                            catalog.warehouse(),
-                            catalog.fileIO(),
-                            options.get(PrivilegedCatalog.USER),
-                            options.get(PrivilegedCatalog.PASSWORD));
+                        options.get(PrivilegedCatalog.PASSWORD));
+        FileBasedPrivilegeManager fileBasedPrivilegeManager =
+                fileBasedPrivilegeManagerLoader.load();
+
+        if (fileBasedPrivilegeManager.privilegeEnabled()) {
+            catalog = new PrivilegedCatalog(catalog, fileBasedPrivilegeManagerLoader);
         }
         return catalog;
     }
@@ -88,8 +77,7 @@ public PrivilegeManager privilegeManager() {
 
     @Override
     public CatalogLoader catalogLoader() {
-        return new PrivilegedCatalogLoader(
-                wrapped.catalogLoader(), warehouse, fileIO, user, password);
+        return new PrivilegedCatalogLoader(wrapped.catalogLoader(), privilegeManagerLoader);
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/privilege/PrivilegedCatalogLoader.java

@@ -20,36 +20,24 @@
 
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.CatalogLoader;
-import org.apache.paimon.fs.FileIO;
 
 /** Loader to create {@link PrivilegedCatalog}. */
 public class PrivilegedCatalogLoader implements CatalogLoader {
 
     private static final long serialVersionUID = 1L;
 
     private final CatalogLoader catalogLoader;
-
-    private final String warehouse;
-    private final FileIO fileIO;
-    private final String user;
-    private final String password;
+    private final PrivilegeManagerLoader privilegeManagerLoader;
 
     public PrivilegedCatalogLoader(
-            CatalogLoader catalogLoader,
-            String warehouse,
-            FileIO fileIO,
-            String user,
-            String password) {
+            CatalogLoader catalogLoader, PrivilegeManagerLoader privilegeManagerLoader) {
         this.catalogLoader = catalogLoader;
-        this.warehouse = warehouse;
-        this.fileIO = fileIO;
-        this.user = user;
-        this.password = password;
+        this.privilegeManagerLoader = privilegeManagerLoader;
     }
 
     @Override
     public Catalog load() {
         Catalog catalog = catalogLoader.load();
-        return new PrivilegedCatalog(catalog, warehouse, fileIO, user, password);
+        return new PrivilegedCatalog(catalog, privilegeManagerLoader);
     }
 }

paimon-core/src/test/java/org/apache/paimon/privilege/PrivilegedCatalogTest.java

@@ -74,7 +74,8 @@ public void testGetTable() throws Exception {
     }
 
     private PrivilegedCatalog create(Catalog catalog, String user, String password) {
-        return new PrivilegedCatalog(catalog, warehouse, fileIO, user, password);
+        return new PrivilegedCatalog(
+                catalog, new FileBasedPrivilegeManagerLoader(warehouse, fileIO, user, password));
     }
 
     private void assertNoPrivilege(Executable executable) {