[server]: prevent delete default database (#3312)

Author: gyang94

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

@@ -196,19 +196,23 @@ void beforeEach() throws Exception {
                         mockLakeCatalog);
         catalog.open();
 
-        // First check if database exists, and drop it if it does
+        // Clean up any leftover tables from previous tests
         if (catalog.databaseExists(DEFAULT_DB)) {
-            catalog.dropDatabase(DEFAULT_DB, true, true);
-        }
-        try {
-            catalog.createDatabase(
-                    DEFAULT_DB, new CatalogDatabaseImpl(Collections.emptyMap(), null), true);
-        } catch (CatalogException e) {
-            // the auto partitioned manager may create the db zk node
-            // in another thread, so if exception is NodeExistsException, just ignore
-            if (!ExceptionUtils.findThrowableWithMessage(e, "KeeperException$NodeExistsException")
-                    .isPresent()) {
-                throw e;
+            for (String table : catalog.listTables(DEFAULT_DB)) {
+                catalog.dropTable(new ObjectPath(DEFAULT_DB, table), true);
+            }
+        } else {
+            try {
+                catalog.createDatabase(
+                        DEFAULT_DB, new CatalogDatabaseImpl(Collections.emptyMap(), null), true);
+            } catch (CatalogException e) {
+                // the auto partitioned manager may create the db zk node
+                // in another thread, so if exception is NodeExistsException, just ignore
+                if (!ExceptionUtils.findThrowableWithMessage(
+                                e, "KeeperException$NodeExistsException")
+                        .isPresent()) {
+                    throw e;
+                }
             }
         }
     }

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

@@ -314,6 +314,12 @@ public Map<String, PartitionRegistration> listPartitions(
 
     public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade)
             throws DatabaseNotExistException, DatabaseNotEmptyException {
+        if (CoordinatorServer.DEFAULT_DATABASE.equals(name)) {
+            throw new UnsupportedOperationException(
+                    "Cannot drop the default database '"
+                            + name
+                            + "'. The default database is required for cluster operation.");
+        }
         if (!databaseExists(name)) {
             if (ignoreIfNotExists) {
                 return;