Throw Exception If catalog default database is not existed or authorized.

Author: loserwang1024

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

@@ -108,7 +108,7 @@ public class FlinkCatalog extends AbstractCatalog {
     protected final ClassLoader classLoader;
 
     protected final String catalogName;
-    protected final @Nullable String defaultDatabase;
+    protected final String defaultDatabase;
     protected final String bootstrapServers;
     private final Map<String, String> securityConfigs;
     protected Connection connection;
@@ -117,7 +117,7 @@ public class FlinkCatalog extends AbstractCatalog {
 
     public FlinkCatalog(
             String name,
-            @Nullable String defaultDatabase,
+            String defaultDatabase,
             String bootstrapServers,
             ClassLoader classLoader,
             Map<String, String> securityConfigs) {
@@ -142,6 +142,10 @@ public void open() throws CatalogException {
 
         connection = ConnectionFactory.createConnection(Configuration.fromMap(flussConfigs));
         admin = connection.getAdmin();
+        if (!databaseExists(defaultDatabase)) {
+            throw new CatalogException(
+                    String.format("Database %s does not exist in fluss server.", defaultDatabase));
+        }
     }
 
     @Override

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

@@ -20,6 +20,7 @@
 import org.apache.fluss.cluster.ServerNode;
 import org.apache.fluss.config.ConfigOptions;
 import org.apache.fluss.config.Configuration;
+import org.apache.fluss.exception.AuthorizationException;
 import org.apache.fluss.exception.InvalidTableException;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.server.testutils.FlussClusterExtension;
@@ -33,6 +34,7 @@
 import org.apache.flink.table.catalog.Catalog;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
 import org.apache.flink.types.Row;
 import org.apache.flink.util.CloseableIterator;
 import org.apache.flink.util.CollectionUtil;
@@ -54,6 +56,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.apache.fluss.config.ConfigOptions.AUTHORIZER_ENABLED;
 import static org.apache.fluss.config.ConfigOptions.DEFAULT_LISTENER_NAME;
 import static org.apache.fluss.flink.FlinkConnectorOptions.BOOTSTRAP_SERVERS;
 import static org.apache.fluss.flink.FlinkConnectorOptions.BUCKET_KEY;
@@ -594,12 +597,13 @@ void testAuthentication() throws Exception {
         Configuration serverConfig = new Configuration();
         serverConfig.setString(ConfigOptions.SERVER_SECURITY_PROTOCOL_MAP.key(), "CLIENT:sasl");
         serverConfig.setString("security.sasl.enabled.mechanisms", "plain");
+        serverConfig.setString(AUTHORIZER_ENABLED.key(), "true");
         serverConfig.setString(
                 "security.sasl.plain.jaas.config",
                 "org.apache.fluss.security.auth.sasl.plain.PlainLoginModule required "
                         + "    user_root=\"password\" "
                         + "    user_guest=\"password2\";");
-        serverConfig.setString(ConfigOptions.SUPER_USERS.key(), "USER:root");
+        serverConfig.setString(ConfigOptions.SUPER_USERS.key(), "User:root");
         FlussClusterExtension flussClusterExtension =
                 FlussClusterExtension.builder()
                         .setCoordinatorServerListeners(
@@ -613,13 +617,34 @@ void testAuthentication() throws Exception {
                         .setClusterConf(serverConfig)
                         .build();
         Catalog authenticateCatalog = null;
+
         try {
             flussClusterExtension.start();
             ServerNode coordinatorServerNode =
                     flussClusterExtension.getCoordinatorServerNode(clientListenerName);
             String bootstrapServers =
                     String.format(
                             "%s:%d", coordinatorServerNode.host(), coordinatorServerNode.port());
+
+            assertThatThrownBy(
+                            () ->
+                                    tEnv.executeSql(
+                                            String.format(
+                                                    "create catalog test_non_authorization_catalog with ('type' = 'fluss', "
+                                                            + "'%s' = '%s', "
+                                                            + "'default-database' = '%s', "
+                                                            + "'client.security.protocol' = 'sasl',"
+                                                            + "'client.security.sasl.username' = 'guest', "
+                                                            + "'client.security.sasl.password' = 'password2' "
+                                                            + " )",
+                                                    BOOTSTRAP_SERVERS.key(),
+                                                    bootstrapServers,
+                                                    DEFAULT_DB)))
+                    .rootCause()
+                    .isExactlyInstanceOf(AuthorizationException.class)
+                    .hasMessageContaining(
+                            "Principal FlussPrincipal{name='guest', type='User'} have no authorization to operate DESCRIBE on resource Resource{type=DATABASE, name='fluss'}");
+
             authenticateCatalog =
                     new FlinkCatalog(
                             CATALOG_NAME,
@@ -635,7 +660,6 @@ void testAuthentication() throws Exception {
 
             Map<String, String> clientConfig = new HashMap<>();
             clientConfig.put(ConfigOptions.CLIENT_SECURITY_PROTOCOL.key(), "sasl");
-            clientConfig.put(ConfigOptions.CLIENT_SASL_MECHANISM.key(), "plain");
             clientConfig.put("client.security.sasl.username", "root");
             clientConfig.put("client.security.sasl.password", "password");
             authenticateCatalog =
@@ -657,6 +681,20 @@ void testAuthentication() throws Exception {
         }
     }
 
+    @Test
+    void createCatalogWithUnexistedDatabase() {
+        assertThatThrownBy(
+                        () ->
+                                tEnv.executeSql(
+                                        String.format(
+                                                "create catalog test_non_exist_database_catalog with ('type' = 'fluss', '%s' = '%s', 'default-database' = 'non-exist')",
+                                                BOOTSTRAP_SERVERS.key(),
+                                                FLUSS_CLUSTER_EXTENSION.getBootstrapServers())))
+                .rootCause()
+                .isExactlyInstanceOf(CatalogException.class)
+                .hasMessage("Database non-exist does not exist in fluss server.");
+    }
+
     private static void assertOptionsEqual(
             Map<String, String> actualOptions, Map<String, String> expectedOptions) {
         actualOptions.remove(ConfigOptions.BOOTSTRAP_SERVERS.key());

fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java

@@ -219,6 +219,13 @@ public CompletableFuture<GetDatabaseInfoResponse> getDatabaseInfo(
 
     @Override
     public CompletableFuture<DatabaseExistsResponse> databaseExists(DatabaseExistsRequest request) {
+        if (authorizer != null) {
+            authorizer.authorize(
+                    currentSession(),
+                    OperationType.DESCRIBE,
+                    Resource.database(request.getDatabaseName()));
+        }
+
         DatabaseExistsResponse response = new DatabaseExistsResponse();
         boolean exists = metadataManager.databaseExists(request.getDatabaseName());
         response.setExists(exists);