[#5066] refactor(drop-catalog): re-define drop catalog (#5067)

### What changes were proposed in this pull request?

- Add an `in-use` property to the catalog with the default value of
true.
 - Only empty catalog with `in-use=false` can be dropped.
 - User can use `dorce` option to drop catalog 
- More drop catalog limitations please see the JavaDoc of `dropCatalog`
in API module

### Why are the changes needed?

Fix: #5066 

### Does this PR introduce _any_ user-facing change?

yes, users now can not drop an in used catalog

### How was this patch tested?

tests added

api/src/main/java/org/apache/gravitino/Catalog.java

@@ -98,6 +98,9 @@ enum CloudName {
    */
   String PROPERTY_PACKAGE = "package";
 
+  /** The property indicates the catalog is in use. */
+  String PROPERTY_IN_USE = "in-use";
+
   /**
    * The property to specify the cloud that the catalog is running on. The value should be one of
    * the {@link CloudName}.

api/src/main/java/org/apache/gravitino/SupportsCatalogs.java

@@ -21,8 +21,11 @@
 import java.util.Map;
 import org.apache.gravitino.annotation.Evolving;
 import org.apache.gravitino.exceptions.CatalogAlreadyExistsException;
+import org.apache.gravitino.exceptions.CatalogInUseException;
+import org.apache.gravitino.exceptions.CatalogNotInUseException;
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
 
 /**
  * Client interface for supporting catalogs. It includes methods for listing, loading, creating,
@@ -48,9 +51,9 @@ public interface SupportsCatalogs {
   Catalog[] listCatalogsInfo() throws NoSuchMetalakeException;
 
   /**
-   * Load a catalog by its identifier.
+   * Load a catalog by its name.
    *
-   * @param catalogName the identifier of the catalog.
+   * @param catalogName the name of the catalog.
    * @return The catalog.
    * @throws NoSuchCatalogException If the catalog does not exist.
    */
@@ -59,7 +62,7 @@ public interface SupportsCatalogs {
   /**
    * Check if a catalog exists.
    *
-   * @param catalogName The identifier of the catalog.
+   * @param catalogName The name of the catalog.
    * @return True if the catalog exists, false otherwise.
    */
   default boolean catalogExists(String catalogName) {
@@ -72,7 +75,7 @@ default boolean catalogExists(String catalogName) {
   }
 
   /**
-   * Create a catalog with specified identifier.
+   * Create a catalog with specified catalog name, type, provider, comment, and properties.
    *
    * <p>The parameter "provider" is a short name of the catalog, used to tell Gravitino which
    * catalog should be created. The short name should be the same as the {@link CatalogProvider}
@@ -96,9 +99,9 @@ Catalog createCatalog(
       throws NoSuchMetalakeException, CatalogAlreadyExistsException;
 
   /**
-   * Alter a catalog with specified identifier.
+   * Alter a catalog with specified catalog name and changes.
    *
-   * @param catalogName the identifier of the catalog.
+   * @param catalogName the name of the catalog.
    * @param changes the changes to apply to the catalog.
    * @return The altered catalog.
    * @throws NoSuchCatalogException If the catalog does not exist.
@@ -108,12 +111,71 @@ Catalog alterCatalog(String catalogName, CatalogChange... changes)
       throws NoSuchCatalogException, IllegalArgumentException;
 
   /**
-   * Drop a catalog with specified identifier.
+   * Drop a catalog with specified name. Please make sure:
+   *
+   * <ul>
+   *   <li>There is no schema in the catalog. Otherwise, a {@link NonEmptyEntityException} will be
+   *       thrown.
+   *   <li>The method {@link #disableCatalog(String)} has been called before dropping the catalog.
+   *       Otherwise, a {@link CatalogInUseException} will be thrown.
+   * </ul>
+   *
+   * It is equivalent to calling {@code dropCatalog(ident, false)}.
    *
    * @param catalogName the name of the catalog.
-   * @return True if the catalog was dropped, false otherwise.
+   * @return True if the catalog was dropped, false if the catalog does not exist.
+   * @throws NonEmptyEntityException If the catalog is not empty.
+   * @throws CatalogInUseException If the catalog is in use.
+   */
+  default boolean dropCatalog(String catalogName)
+      throws NonEmptyEntityException, CatalogInUseException {
+    return dropCatalog(catalogName, false);
+  }
+
+  /**
+   * Drop a catalog with specified name. If the force flag is true, it will:
+   *
+   * <ul>
+   *   <li>Cascade drop all sub-entities (schemas, tables, etc.) of the catalog in Gravitino store.
+   *   <li>Drop the catalog even if it is in use.
+   *   <li>External resources (e.g. database, table, etc.) associated with sub-entities will not be
+   *       dropped unless it is managed (such as managed fileset).
+   * </ul>
+   *
+   * If the force flag is false, it is equivalent to calling {@link #dropCatalog(String)}.
+   *
+   * @param catalogName The identifier of the catalog.
+   * @param force Whether to force the drop.
+   * @return True if the catalog was dropped, false if the catalog does not exist.
+   * @throws NonEmptyEntityException If the catalog is not empty and force is false.
+   * @throws CatalogInUseException If the catalog is in use and force is false.
+   */
+  boolean dropCatalog(String catalogName, boolean force)
+      throws NonEmptyEntityException, CatalogInUseException;
+
+  /**
+   * Enable a catalog. If the catalog is already enabled, this method does nothing.
+   *
+   * @param catalogName The identifier of the catalog.
+   * @throws NoSuchCatalogException If the catalog does not exist.
+   */
+  void enableCatalog(String catalogName) throws NoSuchCatalogException;
+
+  /**
+   * Disable a catalog. If the catalog is already disabled, this method does nothing. Once a catalog
+   * is disabled:
+   *
+   * <ul>
+   *   <li>It can only be listed, loaded, dropped, or disable.
+   *   <li>Any other operations on the catalog will throw an {@link CatalogNotInUseException}.
+   *   <li>Any operation on the sub-entities (schemas, tables, etc.) will throw an {@link
+   *       CatalogNotInUseException}.
+   * </ul>
+   *
+   * @param catalogName The identifier of the catalog.
+   * @throws NoSuchCatalogException If the catalog does not exist.
    */
-  boolean dropCatalog(String catalogName);
+  void disableCatalog(String catalogName) throws NoSuchCatalogException;
 
   /**
    * Test whether the catalog with specified parameters can be connected to before creating it.

api/src/main/java/org/apache/gravitino/exceptions/CatalogInUseException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** Exception thrown when a catalog is in use and cannot be deleted. */
+public class CatalogInUseException extends InUseException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public CatalogInUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

api/src/main/java/org/apache/gravitino/exceptions/CatalogNotInUseException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when operating on a catalog that is not in use. */
+public class CatalogNotInUseException extends NotInUseException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public CatalogNotInUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

api/src/main/java/org/apache/gravitino/exceptions/InUseException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** Exception thrown when a resource is in use and cannot be deleted. */
+public class InUseException extends GravitinoRuntimeException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public InUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

api/src/main/java/org/apache/gravitino/exceptions/NotInUseException.java

@@ -0,0 +1,36 @@
+/*
+ * 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.gravitino.exceptions;
+
+import com.google.errorprone.annotations.FormatMethod;
+import com.google.errorprone.annotations.FormatString;
+
+/** An exception thrown when operating on a resource that is not in use. */
+public class NotInUseException extends GravitinoRuntimeException {
+  /**
+   * Constructs a new exception with the specified detail message.
+   *
+   * @param message the detail message.
+   * @param args the arguments to the message.
+   */
+  @FormatMethod
+  public NotInUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/integration/test/RangerHiveE2EIT.java

@@ -190,10 +190,7 @@ public void stop() throws IOException {
                 catalog.asSchemas().dropSchema(schema, true);
               }));
       Arrays.stream(metalake.listCatalogs())
-          .forEach(
-              (catalogName -> {
-                metalake.dropCatalog(catalogName);
-              }));
+          .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
       client.dropMetalake(metalakeName);
     }
     if (sparkSession != null) {

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/SecureHadoopCatalogOperations.java

@@ -156,6 +156,10 @@ public boolean dropSchema(NameIdentifier ident, boolean cascade) throws NonEmpty
       UserContext.clearUserContext(ident);
 
       return r;
+    } catch (NoSuchEntityException e) {
+      LOG.warn("Schema {} does not exist", ident);
+      return false;
+
     } catch (IOException ioe) {
       throw new RuntimeException("Failed to delete schema " + ident, ioe);
     }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java

@@ -87,7 +87,7 @@ public void setup() throws IOException {
   public void stop() throws IOException {
     Catalog catalog = metalake.loadCatalog(catalogName);
     catalog.asSchemas().dropSchema(schemaName, true);
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
     client.dropMetalake(metalakeName);
     if (fileSystem != null) {
       fileSystem.close();
@@ -160,7 +160,7 @@ void testAlterCatalogLocation() {
 
     Assertions.assertEquals(newLocation, modifiedCatalog.properties().get("location"));
 
-    metalake.dropCatalog(catalogName);
+    metalake.dropCatalog(catalogName, true);
   }
 
   @Test
@@ -606,7 +606,7 @@ public void testDropCatalogWithEmptySchema() {
         filesetCatalog.asSchemas().schemaExists(schemaName), "schema should not be exists");
 
     // Drop the catalog.
-    dropped = metalake.dropCatalog(catalogName);
+    dropped = metalake.dropCatalog(catalogName, true);
     Assertions.assertTrue(dropped, "catalog should be dropped");
     Assertions.assertFalse(metalake.catalogExists(catalogName), "catalog should not be exists");
   }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopUserAuthenticationIT.java

@@ -653,7 +653,7 @@ void testUserImpersonation() {
 
     catalog.asFilesetCatalog().dropFileset(NameIdentifier.of(SCHEMA_NAME, filesetName));
     catalog.asSchemas().dropSchema(SCHEMA_NAME, true);
-    gravitinoMetalake.dropCatalog(catalogName);
+    gravitinoMetalake.dropCatalog(catalogName, true);
     adminClient.dropMetalake(metalakeName);
   }
 }

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/TestHiveCatalogOperations.java

@@ -22,6 +22,7 @@
 import static org.apache.gravitino.Catalog.AUTHORIZATION_PROVIDER;
 import static org.apache.gravitino.Catalog.CLOUD_NAME;
 import static org.apache.gravitino.Catalog.CLOUD_REGION_CODE;
+import static org.apache.gravitino.Catalog.PROPERTY_IN_USE;
 import static org.apache.gravitino.catalog.hive.HiveCatalogPropertiesMeta.CHECK_INTERVAL_SEC;
 import static org.apache.gravitino.catalog.hive.HiveCatalogPropertiesMeta.CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS;
 import static org.apache.gravitino.catalog.hive.HiveCatalogPropertiesMeta.CLIENT_POOL_SIZE;
@@ -73,10 +74,11 @@ void testPropertyMeta() {
     Map<String, PropertyEntry<?>> propertyEntryMap =
         HIVE_PROPERTIES_METADATA.catalogPropertiesMetadata().propertyEntries();
 
-    Assertions.assertEquals(20, propertyEntryMap.size());
+    Assertions.assertEquals(21, propertyEntryMap.size());
     Assertions.assertTrue(propertyEntryMap.containsKey(METASTORE_URIS));
     Assertions.assertTrue(propertyEntryMap.containsKey(Catalog.PROPERTY_PACKAGE));
     Assertions.assertTrue(propertyEntryMap.containsKey(BaseCatalog.CATALOG_OPERATION_IMPL));
+    Assertions.assertTrue(propertyEntryMap.containsKey(PROPERTY_IN_USE));
     Assertions.assertTrue(propertyEntryMap.containsKey(AUTHORIZATION_PROVIDER));
     Assertions.assertTrue(propertyEntryMap.containsKey(CLIENT_POOL_SIZE));
     Assertions.assertTrue(propertyEntryMap.containsKey(IMPERSONATION_ENABLE));

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHiveIT.java

@@ -228,10 +228,7 @@ public void stop() throws IOException {
                 catalog.asSchemas().dropSchema(schema, true);
               }));
       Arrays.stream(metalake.listCatalogs())
-          .forEach(
-              (catalogName -> {
-                metalake.dropCatalog(catalogName);
-              }));
+          .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
       client.dropMetalake(metalakeName);
     }
     if (hiveClientPool != null) {
@@ -1672,7 +1669,7 @@ void testAlterCatalogProperties() {
         });
 
     newCatalog.asSchemas().dropSchema("schema", true);
-    metalake.dropCatalog(nameOfCatalog);
+    metalake.dropCatalog(nameOfCatalog, true);
   }
 
   private void createCatalogWithCustomOperation(String catalogName, String customImpl) {

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/HiveUserAuthenticationIT.java

@@ -269,7 +269,7 @@ public void testUserAuthentication() {
     Assertions.assertFalse(catalog.asSchemas().schemaExists(SCHEMA_NAME));
 
     // Drop catalog
-    Assertions.assertTrue(gravitinoMetalake.dropCatalog(CATALOG_NAME));
+    Assertions.assertTrue(gravitinoMetalake.dropCatalog(CATALOG_NAME, true));
   }
 
   @AfterAll