[#5237][#5240] fix(metalake): fix list metalakes missing in-use prope…

…rty (#5241)

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

 - fix list metalakes missing in-use property
 - fix NPE when null properties

### Why are the changes needed?

list metalakes will show metalake details by default so need put in-use
property in it

Fix: #5240 
Fix: #5237

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

no

### How was this patch tested?

tests added

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

@@ -0,0 +1,37 @@
+/*
+ * 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 not empty. */
+public class NonEmptyCatalogException 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 NonEmptyCatalogException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

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

@@ -0,0 +1,37 @@
+/*
+ * 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 not empty. */
+public class NonEmptyMetalakeException 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 NonEmptyMetalakeException(@FormatString String message, Object... args) {
+    super(message, args);
+  }
+}

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

@@ -21,7 +21,7 @@
 import com.google.errorprone.annotations.FormatMethod;
 import com.google.errorprone.annotations.FormatString;
 
-/** Exception thrown when a namespace is not empty. */
+/** Exception thrown when a schema is not empty. */
 public class NonEmptySchemaException extends GravitinoRuntimeException {
   /**
    * Constructs a new exception with the specified detail message.

clients/client-java/src/main/java/org/apache/gravitino/client/ErrorHandlers.java

@@ -53,6 +53,8 @@
 import org.apache.gravitino.exceptions.NoSuchTagException;
 import org.apache.gravitino.exceptions.NoSuchTopicException;
 import org.apache.gravitino.exceptions.NoSuchUserException;
+import org.apache.gravitino.exceptions.NonEmptyCatalogException;
+import org.apache.gravitino.exceptions.NonEmptyMetalakeException;
 import org.apache.gravitino.exceptions.NonEmptySchemaException;
 import org.apache.gravitino.exceptions.NotFoundException;
 import org.apache.gravitino.exceptions.NotInUseException;
@@ -465,6 +467,9 @@ public void accept(ErrorResponse errorResponse) {
             throw new NotInUseException(errorMessage);
           }
 
+        case ErrorConstants.NON_EMPTY_CODE:
+          throw new NonEmptyCatalogException(errorMessage);
+
         default:
           super.accept(errorResponse);
       }
@@ -496,6 +501,12 @@ public void accept(ErrorResponse errorResponse) {
         case ErrorConstants.IN_USE_CODE:
           throw new MetalakeInUseException(errorMessage);
 
+        case ErrorConstants.NOT_IN_USE_CODE:
+          throw new MetalakeNotInUseException(errorMessage);
+
+        case ErrorConstants.NON_EMPTY_CODE:
+          throw new NonEmptyMetalakeException(errorMessage);
+
         default:
           super.accept(errorResponse);
       }

clients/client-java/src/main/java/org/apache/gravitino/client/GravitinoMetalake.java

@@ -383,7 +383,7 @@ public SupportsRoles supportsRoles() {
     return this;
   }
 
-  /*
+  /**
    * List all the tag names under a metalake.
    *
    * @return A list of the tag names under the current metalake.

clients/client-java/src/test/java/org/apache/gravitino/client/integration/test/MetalakeIT.java

@@ -18,44 +18,54 @@
  */
 package org.apache.gravitino.client.integration.test;
 
+import static org.apache.gravitino.Metalake.PROPERTY_IN_USE;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.google.common.collect.ImmutableMap;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Map;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
 import org.apache.gravitino.MetalakeChange;
 import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.SupportsSchemas;
 import org.apache.gravitino.auth.AuthConstants;
 import org.apache.gravitino.client.GravitinoMetalake;
 import org.apache.gravitino.exceptions.IllegalNameIdentifierException;
 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.NonEmptyMetalakeException;
+import org.apache.gravitino.file.FilesetCatalog;
+import org.apache.gravitino.file.FilesetChange;
 import org.apache.gravitino.integration.test.util.BaseIT;
+import org.apache.gravitino.integration.test.util.GravitinoITUtils;
 import org.apache.gravitino.utils.RandomNameUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.MethodOrderer;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.TestMethodOrder;
 
-@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
 public class MetalakeIT extends BaseIT {
   public static String metalakeNameA = RandomNameUtils.genRandomName("metalakeA");
   public static String metalakeNameB = RandomNameUtils.genRandomName("metalakeB");
 
   @BeforeEach
-  private void start() {
+  public void start() {
     // Just in case clean up is needed due to a test failure
     dropMetalakes();
   }
 
   @AfterEach
-  private void stop() {
+  public void stop() {
     dropMetalakes();
   }
 
@@ -88,6 +98,7 @@ public void testLoadMetalake() {
     client.createMetalake(metalakeNameA, "metalake A comment", Collections.emptyMap());
     GravitinoMetalake metaLakeA = client.loadMetalake(metalakeNameA);
     assertEquals(metaLakeA.name(), metalakeNameA);
+    assertEquals("true", metaLakeA.properties().get(PROPERTY_IN_USE));
 
     // metalake does not exist
     NameIdentifier noexist = NameIdentifier.of(metalakeNameB);
@@ -150,6 +161,7 @@ public void testCreateMetalake() {
     assertEquals(metalakeNameA, metalake.name());
     assertEquals("metalake A comment", metalake.comment());
     assertEquals(AuthConstants.ANONYMOUS_USER, metalake.auditInfo().creator());
+    assertEquals("true", metalake.properties().get(PROPERTY_IN_USE));
 
     // Test metalake name already exists
     Map<String, String> emptyMap = Collections.emptyMap();
@@ -210,11 +222,119 @@ public void testUpdateMetalakeWithNullableComment() {
     assertTrue(client.dropMetalake(metalakeNameA, true));
   }
 
+  @Test
+  public void testMetalakeAvailable() {
+    String metalakeName = RandomNameUtils.genRandomName("test_metalake");
+    client.createMetalake(metalakeName, null, null);
+    // test load metalake
+    GravitinoMetalake metalake = client.loadMetalake(metalakeName);
+    Assertions.assertEquals(metalakeName, metalake.name());
+    Assertions.assertEquals("true", metalake.properties().get(PROPERTY_IN_USE));
+
+    // test list metalakes
+    GravitinoMetalake[] metalakes = client.listMetalakes();
+    Assertions.assertEquals(1, metalakes.length);
+    Assertions.assertEquals(metalakeName, metalakes[0].name());
+    Assertions.assertEquals("true", metalakes[0].properties().get(PROPERTY_IN_USE));
+
+    Exception exception =
+        assertThrows(MetalakeInUseException.class, () -> client.dropMetalake(metalakeName));
+    Assertions.assertTrue(exception.getMessage().contains("please disable it first"));
+
+    // create a catalog under the metalake
+    String catalogName = GravitinoITUtils.genRandomName("test_catalog");
+    Catalog catalog =
+        metalake.createCatalog(
+            catalogName, Catalog.Type.FILESET, "hadoop", "catalog comment", ImmutableMap.of());
+    Assertions.assertEquals("true", catalog.properties().get(Catalog.PROPERTY_IN_USE));
+
+    Assertions.assertDoesNotThrow(() -> client.disableMetalake(metalakeName));
+    GravitinoMetalake loadedMetalake = client.loadMetalake(metalakeName);
+    Assertions.assertEquals("false", loadedMetalake.properties().get(PROPERTY_IN_USE));
+
+    exception =
+        assertThrows(
+            MetalakeNotInUseException.class,
+            () -> client.alterMetalake(metalakeName, MetalakeChange.updateComment("new comment")));
+    Assertions.assertTrue(exception.getMessage().contains("please enable it first"));
+
+    // test catalog operations under non-in-use metalake
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () ->
+            loadedMetalake.createCatalog(
+                catalogName, Catalog.Type.FILESET, "dummy", null, Collections.emptyMap()));
+    Assertions.assertThrows(MetalakeNotInUseException.class, loadedMetalake::listCatalogs);
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> loadedMetalake.loadCatalog(catalogName));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> loadedMetalake.dropCatalog(catalogName));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () -> loadedMetalake.alterCatalog(catalogName, CatalogChange.rename("dummy")));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> loadedMetalake.disableCatalog(catalogName));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> loadedMetalake.enableCatalog(catalogName));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () ->
+            loadedMetalake.testConnection(
+                catalogName, Catalog.Type.FILESET, "dummy", null, Collections.emptyMap()));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> loadedMetalake.catalogExists(catalogName));
+
+    // test schema operations under non-in-use metalake
+    SupportsSchemas schemaOps = catalog.asSchemas();
+    Assertions.assertThrows(MetalakeNotInUseException.class, schemaOps::listSchemas);
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> schemaOps.createSchema("dummy", null, null));
+    Assertions.assertThrows(MetalakeNotInUseException.class, () -> schemaOps.loadSchema("dummy"));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () -> schemaOps.alterSchema("dummy", SchemaChange.removeProperty("dummy")));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> schemaOps.dropSchema("dummy", false));
+
+    // test fileset operations under non-in-use catalog
+    FilesetCatalog filesetOps = catalog.asFilesetCatalog();
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class, () -> filesetOps.listFilesets(Namespace.of("dummy")));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () -> filesetOps.loadFileset(NameIdentifier.of("dummy", "dummy")));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () ->
+            filesetOps.createFileset(NameIdentifier.of("dummy", "dummy"), null, null, null, null));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () -> filesetOps.dropFileset(NameIdentifier.of("dummy", "dummy")));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () -> filesetOps.getFileLocation(NameIdentifier.of("dummy", "dummy"), "dummy"));
+    Assertions.assertThrows(
+        MetalakeNotInUseException.class,
+        () ->
+            filesetOps.alterFileset(
+                NameIdentifier.of("dummy", "dummy"), FilesetChange.removeComment()));
+
+    Assertions.assertThrows(
+        NonEmptyMetalakeException.class, () -> client.dropMetalake(metalakeName));
+
+    Assertions.assertDoesNotThrow(() -> client.enableMetalake(metalakeName));
+    Assertions.assertTrue(loadedMetalake.dropCatalog(catalogName, true));
+
+    Assertions.assertDoesNotThrow(() -> client.disableMetalake(metalakeName));
+    Assertions.assertTrue(client.dropMetalake(metalakeName));
+    Assertions.assertFalse(client.dropMetalake(metalakeName));
+  }
+
   public void dropMetalakes() {
     GravitinoMetalake[] metaLakes = client.listMetalakes();
     for (GravitinoMetalake metalake : metaLakes) {
       assertDoesNotThrow(() -> client.disableMetalake(metalake.name()));
-      assertTrue(client.dropMetalake(metalake.name()));
+      assertTrue(client.dropMetalake(metalake.name(), true));
     }
 
     // Reload metadata from backend to check if the drop operations are applied

common/src/main/java/org/apache/gravitino/dto/requests/MetalakeSetRequest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.gravitino.dto.requests;
 
+import com.fasterxml.jackson.annotation.JsonProperty;
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.ToString;
@@ -29,6 +30,7 @@
 @ToString
 public class MetalakeSetRequest implements RESTRequest {
 
+  @JsonProperty("inUse")
   private final boolean inUse;
 
   /** Default constructor for MetalakeSetRequest. */

core/src/main/java/org/apache/gravitino/catalog/CatalogManager.java

@@ -87,6 +87,7 @@
 import org.apache.gravitino.exceptions.NoSuchCatalogException;
 import org.apache.gravitino.exceptions.NoSuchEntityException;
 import org.apache.gravitino.exceptions.NoSuchMetalakeException;
+import org.apache.gravitino.exceptions.NonEmptyCatalogException;
 import org.apache.gravitino.exceptions.NonEmptyEntityException;
 import org.apache.gravitino.file.FilesetCatalog;
 import org.apache.gravitino.messaging.TopicCatalog;
@@ -629,6 +630,9 @@ public Catalog alterCatalog(NameIdentifier ident, CatalogChange... changes)
   @Override
   public boolean dropCatalog(NameIdentifier ident, boolean force)
       throws NonEmptyEntityException, CatalogInUseException {
+    NameIdentifier metalakeIdent = NameIdentifier.of(ident.namespace().levels());
+    checkMetalake(metalakeIdent, store);
+
     try {
       boolean catalogInUse = catalogInUse(store, ident);
       if (catalogInUse && !force) {
@@ -646,7 +650,7 @@ public boolean dropCatalog(NameIdentifier ident, boolean force)
       if (!schemas.isEmpty() && !force) {
         // the Kafka catalog is special, it includes a default schema
         if (!catalogEntity.getProvider().equals("kafka") || schemas.size() > 1) {
-          throw new NonEmptyEntityException(
+          throw new NonEmptyCatalogException(
               "Catalog %s has schemas, please drop them first or use force option", ident);
         }
       }