diff --git a/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java b/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
index 8515ea7d20..30e72c0413 100644
--- a/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
+++ b/catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogOperations.java
@@ -151,7 +151,9 @@ public void initialize(
                 .getOrDefault(config, HadoopCatalogPropertiesMetadata.LOCATION);
     this.catalogStorageLocation =
         StringUtils.isNotBlank(catalogLocation)
-            ? Optional.of(catalogLocation).map(Path::new)
+            ? Optional.of(catalogLocation)
+                .map(s -> s.endsWith(SLASH) ? s : s + SLASH)
+                .map(Path::new)
             : Optional.empty();
   }
 
@@ -722,6 +724,7 @@ private Path getSchemaPath(String name, Map<String, String> properties) {
                 .getOrDefault(properties, HadoopSchemaPropertiesMetadata.LOCATION);
 
     return Optional.ofNullable(schemaLocation)
+        .map(s -> s.endsWith(SLASH) ? s : s + SLASH)
         .map(Path::new)
         .orElse(catalogStorageLocation.map(p -> new Path(p, name)).orElse(null));
   }
diff --git a/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java b/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
index 6cd10cbf24..b068e6130d 100644
--- a/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
+++ b/catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java
@@ -690,6 +690,66 @@ public void testGetFileLocationWithInvalidAuditHeaders() {
     }
   }
 
+  @Test
+  public void testCreateSchemaAndFilesetWithSpecialLocation() {
+    String localCatalogName = GravitinoITUtils.genRandomName("local_catalog");
+    String hdfsLocation =
+        String.format(
+            "hdfs://%s:%d",
+            containerSuite.getHiveContainer().getContainerIpAddress(),
+            HiveContainer.HDFS_DEFAULTFS_PORT);
+    Map<String, String> catalogProps = ImmutableMap.of("location", hdfsLocation);
+
+    Catalog localCatalog =
+        metalake.createCatalog(
+            localCatalogName, Catalog.Type.FILESET, provider, "comment", catalogProps);
+    Assertions.assertEquals(hdfsLocation, localCatalog.properties().get("location"));
+
+    // Create schema without specifying location.
+    Schema localSchema =
+        localCatalog
+            .asSchemas()
+            .createSchema("local_schema", "comment", ImmutableMap.of("key1", "val1"));
+
+    Fileset localFileset =
+        localCatalog
+            .asFilesetCatalog()
+            .createFileset(
+                NameIdentifier.of(localSchema.name(), "local_fileset"),
+                "fileset comment",
+                Fileset.Type.MANAGED,
+                null,
+                ImmutableMap.of("k1", "v1"));
+    Assertions.assertEquals(
+        hdfsLocation + "/local_schema/local_fileset", localFileset.storageLocation());
+
+    // Delete schema
+    localCatalog.asSchemas().dropSchema(localSchema.name(), true);
+
+    // Create schema with specifying location.
+    Map<String, String> schemaProps = ImmutableMap.of("location", hdfsLocation);
+    Schema localSchema2 =
+        localCatalog.asSchemas().createSchema("local_schema2", "comment", schemaProps);
+    Assertions.assertEquals(hdfsLocation, localSchema2.properties().get("location"));
+
+    Fileset localFileset2 =
+        localCatalog
+            .asFilesetCatalog()
+            .createFileset(
+                NameIdentifier.of(localSchema2.name(), "local_fileset2"),
+                "fileset comment",
+                Fileset.Type.MANAGED,
+                null,
+                ImmutableMap.of("k1", "v1"));
+    Assertions.assertEquals(hdfsLocation + "/local_fileset2", localFileset2.storageLocation());
+
+    // Delete schema
+    localCatalog.asSchemas().dropSchema(localSchema2.name(), true);
+
+    // Delete catalog
+    metalake.dropCatalog(localCatalogName, true);
+  }
+
   protected String generateLocation(String filesetName) {
     return String.format(
         "hdfs://%s:%d/user/hadoop/%s/%s/%s",