[#7190] fix(catalogs): Validate and filter empty storageLocations in HadoopCatalogOperations

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

@@ -250,6 +250,10 @@ public Fileset createMultipleLocationFileset(
           if (StringUtils.isBlank(name)) {
             throw new IllegalArgumentException("Location name must not be blank");
           }
+          if (StringUtils.isBlank(path)) {
+            throw new IllegalArgumentException(
+                "Storage location must not be blank for location name: " + name);
+          }
         });
 
     // Check if the fileset already existed in cache first. If it does, it means the fileset is
@@ -974,7 +978,19 @@ private void validateLocationHierarchy(
                 }));
   }
 
+  private boolean existBlankValue(Map<String, String> properties, String key) {
+    return properties.containsKey(key) && StringUtils.isBlank(properties.get(key));
+  }
+
   private Map<String, Path> getAndCheckCatalogStorageLocations(Map<String, String> properties) {
+    if (existBlankValue(properties, HadoopCatalogPropertiesMetadata.LOCATION)) {
+      // If the properties contain the catalog property "location", it must not be blank.
+      throw new IllegalArgumentException(
+          "The value of the catalog property "
+              + HadoopCatalogPropertiesMetadata.LOCATION
+              + " must not be blank");
+    }
+
     ImmutableMap.Builder<String, Path> catalogStorageLocations = ImmutableMap.builder();
     String unnamedLocation =
         (String)
@@ -989,11 +1005,17 @@ private Map<String, Path> getAndCheckCatalogStorageLocations(Map<String, String>
 
     properties.forEach(
         (k, v) -> {
-          if (k.startsWith(PROPERTY_MULTIPLE_LOCATIONS_PREFIX) && StringUtils.isNotBlank(v)) {
+          if (k.startsWith(PROPERTY_MULTIPLE_LOCATIONS_PREFIX)) {
             String locationName = k.substring(PROPERTY_MULTIPLE_LOCATIONS_PREFIX.length());
             if (StringUtils.isBlank(locationName)) {
               throw new IllegalArgumentException("Location name must not be blank");
             }
+
+            if (StringUtils.isBlank(v)) {
+              throw new IllegalArgumentException(
+                  "Location value must not be blank for " + "location name: " + locationName);
+            }
+
             checkPlaceholderValue(v);
             catalogStorageLocations.put(locationName, new Path((ensureTrailingSlash(v))));
           }
@@ -1003,6 +1025,14 @@ private Map<String, Path> getAndCheckCatalogStorageLocations(Map<String, String>
 
   private Map<String, Path> getAndCheckSchemaPaths(
       String schemaName, Map<String, String> schemaProps) {
+    if (existBlankValue(schemaProps, HadoopSchemaPropertiesMetadata.LOCATION)) {
+      // If the properties contain the schema property "location", it must not be blank.
+      throw new IllegalArgumentException(
+          "The value of the schema property "
+              + HadoopSchemaPropertiesMetadata.LOCATION
+              + " must not be blank");
+    }
+
     Map<String, Path> schemaPaths = new HashMap<>();
     catalogStorageLocations.forEach(
         (name, path) -> {

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

@@ -50,6 +50,7 @@
 import java.net.ConnectException;
 import java.nio.file.Paths;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
@@ -382,15 +383,15 @@ public void testCreateSchemaWithEmptyCatalogLocation() throws IOException {
     String name = "schema28";
     String comment = "comment28";
     String catalogPath = "";
-    Schema schema = createSchema(name, comment, catalogPath, null);
-    Assertions.assertEquals(name, schema.name());
-    Assertions.assertEquals(comment, schema.comment());
 
     Throwable exception =
         Assertions.assertThrows(
-            SchemaAlreadyExistsException.class,
-            () -> createSchema(name, comment, catalogPath, null));
-    Assertions.assertEquals("Schema m1.c1.schema28 already exists", exception.getMessage());
+            IllegalArgumentException.class, () -> createSchema(name, comment, catalogPath, null));
+    Assertions.assertEquals(
+        "The value of the catalog property "
+            + HadoopCatalogPropertiesMetadata.LOCATION
+            + " must not be blank",
+        exception.getMessage());
   }
 
   @Test
@@ -1493,6 +1494,24 @@ public void testCreateMultipleLocationsWithExceptions() throws IOException {
                       null));
       Assertions.assertEquals("Location name must not be blank", exception.getMessage());
 
+      // empty location in catalog location
+      Map<String, String> illegalLocations2 =
+          new HashMap<String, String>() {
+            {
+              put(PROPERTY_MULTIPLE_LOCATIONS_PREFIX + "v1", null);
+            }
+          };
+      exception =
+          Assertions.assertThrows(
+              IllegalArgumentException.class,
+              () ->
+                  ops.initialize(
+                      illegalLocations2,
+                      randomCatalogInfo("m1", "c1"),
+                      HADOOP_PROPERTIES_METADATA));
+      Assertions.assertEquals(
+          "Location value must not be blank for location name: v1", exception.getMessage());
+
       // storage location is parent of schema location
       Schema multipLocationSchema =
           createMultiLocationSchema(