[#5154] refactor(drop-metalake): re-define drop metalake (#5155)

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

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

### Why are the changes needed?

Fix: #5154 

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

yes, users now can not drop an in used metalake

### How was this patch tested?

tests added

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

@@ -29,6 +29,9 @@
 @Evolving
 public interface Metalake extends Auditable {
 
+  /** The property indicating the metalake is in use. */
+  String PROPERTY_IN_USE = "in-use";
+
   /**
    * The name of the metalake.
    *

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

@@ -21,7 +21,10 @@
 import java.util.Map;
 import org.apache.gravitino.annotation.Evolving;
 import org.apache.gravitino.exceptions.MetalakeAlreadyExistsException;
+import org.apache.gravitino.exceptions.MetalakeInUseException;
+import org.apache.gravitino.exceptions.MetalakeNotInUseException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyEntityException;
 
 /**
  * Client interface for supporting metalakes. It includes methods for listing, loading, creating,
@@ -38,7 +41,7 @@ public interface SupportsMetalakes {
   Metalake[] listMetalakes();
 
   /**
-   * Load a metalake by its identifier.
+   * Load a metalake by its name.
    *
    * @param name the name of the metalake.
    * @return The metalake.
@@ -62,7 +65,7 @@ default boolean metalakeExists(String name) {
   }
 
   /**
-   * Create a metalake with specified identifier.
+   * Create a metalake with specified name, comment and properties.
    *
    * @param name The name of the metalake.
    * @param comment The comment of the metalake.
@@ -74,7 +77,7 @@ Metalake createMetalake(String name, String comment, Map<String, String> propert
       throws MetalakeAlreadyExistsException;
 
   /**
-   * Alter a metalake with specified identifier.
+   * Alter a metalake with specified metalake name and changes.
    *
    * @param name The name of the metalake.
    * @param changes The changes to apply.
@@ -86,10 +89,69 @@ Metalake alterMetalake(String name, MetalakeChange... changes)
       throws NoSuchMetalakeException, IllegalArgumentException;
 
   /**
-   * Drop a metalake with specified identifier.
+   * Drop a metalake with specified name. Please make sure:
    *
-   * @param name The identifier of the metalake.
+   * <ul>
+   *   <li>There is no catalog in the metalake. Otherwise, a {@link NonEmptyEntityException} will be
+   *       thrown.
+   *   <li>The method {@link #disableMetalake(String)} has been called before dropping the metalake.
+   *       Otherwise, a {@link MetalakeInUseException} will be thrown.
+   * </ul>
+   *
+   * It is equivalent to calling {@code dropMetalake(name, false)}.
+   *
+   * @param name The name of the metalake.
    * @return True if the metalake was dropped, false if the metalake does not exist.
+   * @throws NonEmptyEntityException If the metalake is not empty.
+   * @throws MetalakeInUseException If the metalake is in use.
+   */
+  default boolean dropMetalake(String name) throws NonEmptyEntityException, MetalakeInUseException {
+    return dropMetalake(name, false);
+  }
+
+  /**
+   * Drop a metalake with specified name. If the force flag is true, it will:
+   *
+   * <ul>
+   *   <li>Cascade drop all sub-entities (tags, catalogs, schemas, tables, etc.) of the metalake in
+   *       Gravitino store.
+   *   <li>Drop the metalake even if it is in use.
+   *   <li>External resources (e.g. database, table, etc.) associated with sub-entities will not be
+   *       deleted unless it is managed (such as managed fileset).
+   * </ul>
+   *
+   * If the force flag is false, it is equivalent to calling {@link #dropMetalake(String)}.
+   *
+   * @param name The name of the metalake.
+   * @param force Whether to force the drop.
+   * @return True if the metalake was dropped, false if the metalake does not exist.
+   * @throws NonEmptyEntityException If the metalake is not empty and force is false.
+   * @throws MetalakeInUseException If the metalake is in use and force is false.
+   */
+  boolean dropMetalake(String name, boolean force)
+      throws NonEmptyEntityException, MetalakeInUseException;
+
+  /**
+   * Enable a metalake. If the metalake is already in use, this method does nothing.
+   *
+   * @param name The name of the metalake.
+   * @throws NoSuchMetalakeException If the metalake does not exist.
+   */
+  void enableMetalake(String name) throws NoSuchMetalakeException;
+
+  /**
+   * Disable a metalake. If the metalake is already disabled, this method does nothing. Once a
+   * metalake is disable:
+   *
+   * <ul>
+   *   <li>It can only be listed, loaded, dropped, or enable.
+   *   <li>Any other operations on the metalake will throw an {@link MetalakeNotInUseException}.
+   *   <li>Any operation on the sub-entities (catalogs, schemas, tables, etc.) will throw an {@link
+   *       MetalakeNotInUseException}.
+   * </ul>
+   *
+   * @param name The name of the metalake.
+   * @throws NoSuchMetalakeException If the metalake does not exist.
    */
-  boolean dropMetalake(String name);
+  void disableMetalake(String name) throws NoSuchMetalakeException;
 }

api/src/main/java/org/apache/gravitino/exceptions/MetalakeInUseException.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 metalake is in use and cannot be deleted. */
+public class MetalakeInUseException 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 MetalakeInUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

api/src/main/java/org/apache/gravitino/exceptions/MetalakeNotInUseException.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 metalake that is not in use. */
+public class MetalakeNotInUseException 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 MetalakeNotInUseException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

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

@@ -188,6 +188,7 @@ public void stop() {
               }));
       Arrays.stream(metalake.listCatalogs())
           .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
+      client.disableMetalake(metalakeName);
       client.dropMetalake(metalakeName);
     }
     if (sparkSession != null) {
@@ -267,10 +268,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

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

@@ -88,7 +88,7 @@ public void stop() throws IOException {
     Catalog catalog = metalake.loadCatalog(catalogName);
     catalog.asSchemas().dropSchema(schemaName, true);
     metalake.dropCatalog(catalogName, true);
-    client.dropMetalake(metalakeName);
+    client.dropMetalake(metalakeName, true);
     if (fileSystem != null) {
       fileSystem.close();
     }
@@ -104,10 +104,9 @@ protected void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

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

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

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

@@ -258,10 +258,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

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

@@ -229,7 +229,7 @@ public void stop() throws IOException {
               }));
       Arrays.stream(metalake.listCatalogs())
           .forEach((catalogName -> metalake.dropCatalog(catalogName, true)));
-      client.dropMetalake(metalakeName);
+      client.dropMetalake(metalakeName, true);
     }
     if (hiveClientPool != null) {
       hiveClientPool.close();
@@ -264,10 +264,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }
@@ -1429,8 +1428,8 @@ void testDropAndRename() {
     client.createMetalake(metalakeName1, "comment", Collections.emptyMap());
     client.createMetalake(metalakeName2, "comment", Collections.emptyMap());
 
-    client.dropMetalake(metalakeName1);
-    client.dropMetalake(metalakeName2);
+    client.dropMetalake(metalakeName1, true);
+    client.dropMetalake(metalakeName2, true);
 
     client.createMetalake(metalakeName1, "comment", Collections.emptyMap());
 

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

@@ -389,10 +389,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(METALAKE_NAME, "comment", Collections.emptyMap());
+    client.createMetalake(METALAKE_NAME, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(METALAKE_NAME);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(METALAKE_NAME, loadMetalake.name());
 
     metalake = loadMetalake;
   }

catalogs/catalog-jdbc-doris/src/test/java/org/apache/gravitino/catalog/doris/integration/test/CatalogDorisIT.java

@@ -126,7 +126,7 @@ public void startup() throws IOException {
   public void stop() {
     clearTableAndSchema();
     metalake.dropCatalog(catalogName, true);
-    client.dropMetalake(metalakeName);
+    client.dropMetalake(metalakeName, true);
   }
 
   @AfterEach
@@ -143,10 +143,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetaLakes = client.listMetalakes();
     assertEquals(0, gravitinoMetaLakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    assertEquals(createdMetalake, loadMetalake);
+    assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/AuditCatalogMysqlIT.java

@@ -77,7 +77,7 @@ public void startIntegrationTest() throws Exception {
 
   @AfterAll
   public void stopIntegrationTest() throws IOException, InterruptedException {
-    client.dropMetalake(metalakeName);
+    client.dropMetalake(metalakeName, true);
     mysqlService.close();
     super.stopIntegrationTest();
   }
@@ -169,10 +169,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
     metalake = loadMetalake;
   }
 }

catalogs/catalog-jdbc-mysql/src/test/java/org/apache/gravitino/catalog/mysql/integration/test/CatalogMysqlIT.java

@@ -144,6 +144,7 @@ public void stop() {
     clearTableAndSchema();
     metalake.disableCatalog(catalogName);
     metalake.dropCatalog(catalogName);
+    client.disableMetalake(metalakeName);
     client.dropMetalake(metalakeName);
     mysqlService.close();
   }
@@ -167,10 +168,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }

catalogs/catalog-jdbc-postgresql/src/test/java/org/apache/gravitino/catalog/postgresql/integration/test/CatalogPostgreSqlIT.java

@@ -130,6 +130,7 @@ public void stop() {
     }
     metalake.disableCatalog(catalogName);
     metalake.dropCatalog(catalogName);
+    client.disableMetalake(metalakeName);
     client.dropMetalake(metalakeName);
     postgreSqlService.close();
   }
@@ -153,10 +154,9 @@ private void createMetalake() {
     GravitinoMetalake[] gravitinoMetalakes = client.listMetalakes();
     Assertions.assertEquals(0, gravitinoMetalakes.length);
 
-    GravitinoMetalake createdMetalake =
-        client.createMetalake(metalakeName, "comment", Collections.emptyMap());
+    client.createMetalake(metalakeName, "comment", Collections.emptyMap());
     GravitinoMetalake loadMetalake = client.loadMetalake(metalakeName);
-    Assertions.assertEquals(createdMetalake, loadMetalake);
+    Assertions.assertEquals(metalakeName, loadMetalake.name());
 
     metalake = loadMetalake;
   }